wuyang/zk-data-agent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Here's a summary of all the features I ported from the TypeScript source to Python:

Browse Source
This commit is contained in:
Abdelrahman Elsayed Mahmoud Abdallah
2026-04-15 02:29:22 +02:00
parent 18530f5435
commit a4aaf814d7
8 changed files with 1343 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/agent_runtime.py
+276 -33
View File
@@ -65,6 +65,15 @@ from .session_store import (
usage_from_payload,
)
from .token_budget import calculate_token_budget, format_token_budget
from .builtin_agents import (
AgentDefinition,
get_agent_definition,
get_builtin_agents,
format_agent_listing,
ALL_AGENT_DISALLOWED_TOOLS,
GENERAL_PURPOSE_AGENT,
)
from .microcompact import microcompact_messages as _microcompact_messages
@dataclass(frozen=True)
@@ -113,6 +122,7 @@ class LocalCodingAgent:
last_run_result: AgentRunResult | None = field(default=None, init=False, repr=False)
cumulative_usage: UsageStats = field(default_factory=UsageStats, init=False, repr=False)
cumulative_cost_usd: float = field(default=0.0, init=False, repr=False)
_compact_consecutive_failures: int = field(default=0, init=False, repr=False)
active_session_id: str | None = field(default=None, init=False, repr=False)
last_session_path: str | None = field(default=None, init=False, repr=False)
managed_agent_id: str | None = field(default=None, init=False, repr=False)
@@ -465,7 +475,7 @@ class LocalCodingAgent:
stream_events: list[dict[str, object]] = []
assistant_response_segments: list[str] = []
delegated_tasks = sum(
1 for entry in file_history if entry.get('action') == 'delegate_agent'
1 for entry in file_history if entry.get('action') in ('delegate_agent', 'Agent')
)
model_calls = starting_model_calls
@@ -497,6 +507,11 @@ class LocalCodingAgent:
return result
for turn_index in range(1, self.runtime_config.max_turns + 1):
self._microcompact_session_if_needed(
session,
stream_events,
turn_index=turn_index,
)
self._snip_session_if_needed(
session,
stream_events,
@@ -775,7 +790,7 @@ class LocalCodingAgent:
for tool_call in turn.tool_calls:
assistant_response_segments.clear()
tool_calls += 1
if tool_call.name == 'delegate_agent':
if tool_call.name in ('Agent', 'delegate_agent'):
delegated_tasks += self._delegated_task_units(tool_call.arguments)
budget_after_tool_request = self._check_budget(
total_usage,
@@ -907,9 +922,12 @@ class LocalCodingAgent:
'message': policy_block_message,
}
)
if tool_call.name == 'delegate_agent':
if tool_call.name in ('Agent', 'delegate_agent'):
if tool_result is None:
tool_result = self._execute_delegate_agent(tool_call.arguments)
elif tool_call.name == 'Skill':
if tool_result is None:
tool_result = self._execute_skill(tool_call.arguments)
elif tool_result is None:
for update in execute_tool_streaming(
self.tool_registry,
@@ -1365,7 +1383,17 @@ class LocalCodingAgent:
return PromptPreflightResult()
snapshot = recovered
if self._can_auto_compact_with_summary(session):
# Circuit-breaker: skip auto-compact after MAX_COMPACT_FAILURES consecutive failures
from .compact import MAX_COMPACT_FAILURES
if self._compact_consecutive_failures >= MAX_COMPACT_FAILURES:
stream_events.append(
{
'type': 'auto_compact_circuit_breaker',
'turn_index': turn_index,
'consecutive_failures': self._compact_consecutive_failures,
}
)
elif self._can_auto_compact_with_summary(session):
compact_result = compact_conversation(
self,
custom_instructions=(
@@ -1375,6 +1403,7 @@ class LocalCodingAgent:
),
)
if compact_result.error is None:
self._compact_consecutive_failures = 0 # Reset on success
recovered = calculate_token_budget(
session=session,
model=self.model_config.model,
@@ -1387,7 +1416,9 @@ class LocalCodingAgent:
'turn_index': turn_index,
'pre_compact_token_count': compact_result.pre_compact_token_count,
'post_compact_token_count': compact_result.post_compact_token_count,
'true_post_compact_token_count': compact_result.true_post_compact_token_count,
'summary_usage_tokens': compact_result.usage.total_tokens,
'ptl_retries': compact_result.ptl_retries,
'projected_input_tokens': recovered.projected_input_tokens,
'soft_input_limit_tokens': recovered.soft_input_limit_tokens,
'hard_input_limit_tokens': recovered.hard_input_limit_tokens,
@@ -1417,11 +1448,13 @@ class LocalCodingAgent:
),
)
else:
self._compact_consecutive_failures += 1
stream_events.append(
{
'type': 'auto_compact_failed',
'turn_index': turn_index,
'reason': compact_result.error,
'consecutive_failures': self._compact_consecutive_failures,
}
)
@@ -1457,6 +1490,39 @@ class LocalCodingAgent:
f'soft input limit: {snapshot.soft_input_limit_tokens:,}.'
)
def _microcompact_session_if_needed(
self,
session: AgentSessionState,
stream_events: list[dict[str, object]],
*,
turn_index: int,
) -> None:
"""Run time-based microcompaction to clear old tool results.
Fires when the gap since the last assistant message exceeds the
threshold (default 60 minutes), indicating the server-side cache
has expired and the full prefix will be rewritten anyway.
"""
if not session.messages:
return
result = _microcompact_messages(
session.messages,
model=self.model_config.model,
)
if not result.triggered:
return
session.messages = result.messages
stream_events.append(
{
'type': 'microcompact',
'turn_index': turn_index,
'cleared_tool_count': result.cleared_tool_count,
'kept_tool_count': result.kept_tool_count,
'estimated_tokens_saved': result.estimated_tokens_saved,
'gap_minutes': round(result.gap_minutes, 1),
}
)
def _snip_session_if_needed(
self,
session: AgentSessionState,
@@ -1739,7 +1805,7 @@ class LocalCodingAgent:
if (
'path' not in tool_result.metadata
and 'command' not in tool_result.metadata
and tool_result.metadata.get('action') != 'delegate_agent'
and tool_result.metadata.get('action') not in ('delegate_agent', 'Agent')
):
return None
metadata = dict(tool_result.metadata)
@@ -1766,7 +1832,7 @@ class LocalCodingAgent:
entry['after_snapshot_id'] = f'{path}:{after_sha256[:12]}'
elif isinstance(metadata.get('command'), str):
entry['history_kind'] = 'shell'
elif action == 'delegate_agent':
elif action in ('delegate_agent', 'Agent'):
entry['history_kind'] = 'delegation'
delegate_batches = metadata.get('delegate_batches')
if isinstance(delegate_batches, list):
@@ -1972,46 +2038,205 @@ class LocalCodingAgent:
)
return any(pattern in text for pattern in patterns)
def _execute_skill(
self,
arguments: dict[str, object],
) -> ToolExecutionResult:
"""Execute a skill (slash command) through the Skill tool.
Maps the ``skill`` parameter to a slash command, invokes it, and
returns its output as a tool result.
"""
from .agent_slash_commands import find_slash_command, get_slash_command_specs
skill_name = arguments.get('skill')
if not isinstance(skill_name, str) or not skill_name.strip():
return ToolExecutionResult(
name='Skill',
ok=False,
content='skill must be a non-empty string',
)
# Normalize: strip leading '/' if present
skill_name = skill_name.strip().lstrip('/')
args = arguments.get('args', '')
if not isinstance(args, str):
args = str(args) if args is not None else ''
# Look up the slash command
spec = find_slash_command(skill_name)
if spec is None:
available = sorted(
name
for spec in get_slash_command_specs()
for name in spec.names
)
return ToolExecutionResult(
name='Skill',
ok=False,
content=(
f'Unknown skill: {skill_name}. '
f'Available skills: {", ".join(available[:30])}'
),
metadata={'action': 'skill_not_found', 'skill_name': skill_name},
)
# Invoke the slash command handler
input_text = f'/{skill_name} {args}'.strip()
result = spec.handler(self, args.strip(), input_text)
if result.output:
content = result.output
elif result.prompt:
content = result.prompt
else:
content = f'Skill /{skill_name} completed.'
return ToolExecutionResult(
name='Skill',
ok=True,
content=content,
metadata={
'action': 'skill',
'skill_name': skill_name,
'command_name': spec.names[0],
'handled': result.handled,
'should_query': result.should_query,
},
)
def _resolve_agent_definition(
self,
arguments: dict[str, object],
) -> AgentDefinition:
"""Resolve the agent definition from subagent_type or default to general-purpose."""
subagent_type = arguments.get('subagent_type')
if isinstance(subagent_type, str) and subagent_type:
agent_def = get_agent_definition(subagent_type)
if agent_def is not None:
return agent_def
return GENERAL_PURPOSE_AGENT
def _resolve_child_model_config(
self,
arguments: dict[str, object],
agent_def: AgentDefinition,
) -> ModelConfig:
"""Resolve model config for a child agent based on explicit override or agent definition."""
model_override = arguments.get('model')
agent_model = agent_def.model
# Explicit model param in arguments takes priority
if isinstance(model_override, str) and model_override in ('sonnet', 'opus', 'haiku'):
# Map friendly names to actual model identifiers if a mapping is available,
# otherwise use the parent's model config as base
return self.model_config
# Agent definition model
if agent_model and agent_model != 'inherit':
# Agent definitions may specify 'haiku', 'sonnet', 'opus'
# For now, inherit the parent's model config (model routing would
# require a model registry which is out of scope)
return self.model_config
return self.model_config
def _filter_tools_for_agent(
self,
agent_def: AgentDefinition,
) -> dict[str, AgentTool]:
"""Build the tool registry for a child agent based on its definition."""
# Start from parent tools, remove Agent/delegate_agent to prevent recursive spawning
base_tools = {
name: tool
for name, tool in self.tool_registry.items()
if name not in ('delegate_agent', 'Agent')
}
# Apply agent-specific tool allow-list
if agent_def.tools is not None:
allowed = set(agent_def.tools)
base_tools = {
name: tool
for name, tool in base_tools.items()
if name in allowed
}
# Apply agent-specific disallowed tools
if agent_def.disallowed_tools:
denied = set(agent_def.disallowed_tools)
base_tools = {
name: tool
for name, tool in base_tools.items()
if name not in denied
}
# Apply universal agent disallowed tools
base_tools = {
name: tool
for name, tool in base_tools.items()
if name not in ALL_AGENT_DISALLOWED_TOOLS
}
return base_tools
def _execute_delegate_agent(
self,
arguments: dict[str, object],
) -> ToolExecutionResult:
tool_name = 'Agent'
agent_def = self._resolve_agent_definition(arguments)
max_turns = arguments.get('max_turns')
if max_turns is not None and (isinstance(max_turns, bool) or not isinstance(max_turns, int) or max_turns < 1):
return ToolExecutionResult(
name='delegate_agent',
name=tool_name,
ok=False,
content='max_turns must be an integer >= 1',
)
subtasks = self._normalize_delegate_subtasks(arguments)
if not subtasks:
return ToolExecutionResult(
name='delegate_agent',
name=tool_name,
ok=False,
content='prompt must be a non-empty string or subtasks must contain at least one prompt',
)
child_permissions = AgentPermissions(
allow_file_write=(
self.runtime_config.permissions.allow_file_write
and bool(arguments.get('allow_write', False))
),
allow_shell_commands=(
self.runtime_config.permissions.allow_shell_commands
and bool(arguments.get('allow_shell', False))
),
allow_destructive_shell_commands=False,
)
# Resolve child permissions — read-only agents get no write/shell
if agent_def.disallowed_tools and (
'edit_file' in agent_def.disallowed_tools
or 'write_file' in agent_def.disallowed_tools
):
# Read-only agent (Explore, Plan, verification)
child_permissions = AgentPermissions(
allow_file_write=False,
allow_shell_commands=self.runtime_config.permissions.allow_shell_commands,
allow_destructive_shell_commands=False,
)
else:
child_permissions = AgentPermissions(
allow_file_write=(
self.runtime_config.permissions.allow_file_write
and bool(arguments.get('allow_write', False))
),
allow_shell_commands=(
self.runtime_config.permissions.allow_shell_commands
and bool(arguments.get('allow_shell', False))
),
allow_destructive_shell_commands=False,
)
# Resolve max_turns — agent definition or explicit param
effective_max_turns = max_turns or agent_def.max_turns or min(self.runtime_config.max_turns, 6)
child_runtime_config = replace(
self.runtime_config,
max_turns=max_turns or min(self.runtime_config.max_turns, 6),
max_turns=effective_max_turns,
permissions=child_permissions,
auto_compact_threshold_tokens=self.runtime_config.auto_compact_threshold_tokens,
)
child_tools = {
name: tool
for name, tool in self.tool_registry.items()
if name != 'delegate_agent'
}
child_model_config = self._resolve_child_model_config(arguments, agent_def)
child_tools = self._filter_tools_for_agent(agent_def)
include_parent_context = bool(arguments.get('include_parent_context', True))
continue_on_error = bool(arguments.get('continue_on_error', True))
max_failures = arguments.get('max_failures')
@@ -2112,15 +2337,32 @@ class LocalCodingAgent:
stop_processing = True
break
continue
# Use agent definition's system prompt if available
child_system_prompt = agent_def.system_prompt or self.custom_system_prompt
child_override_prompt = None
if agent_def.system_prompt:
child_override_prompt = agent_def.system_prompt
else:
child_override_prompt = self.override_system_prompt
# Inject critical system reminder if agent definition has one
child_append_prompt = self.append_system_prompt
if agent_def.critical_system_reminder:
reminder = f'\n\n<system-reminder>\n{agent_def.critical_system_reminder}\n</system-reminder>'
child_append_prompt = (
(child_append_prompt or '') + reminder
)
child_agent = LocalCodingAgent(
model_config=self.model_config,
model_config=child_model_config,
runtime_config=replace(
child_runtime_config,
max_turns=subtask.get('max_turns', child_runtime_config.max_turns),
disable_claude_md_discovery=agent_def.omit_claude_md,
),
custom_system_prompt=self.custom_system_prompt,
append_system_prompt=self.append_system_prompt,
override_system_prompt=self.override_system_prompt,
custom_system_prompt=child_system_prompt if not child_override_prompt else None,
append_system_prompt=child_append_prompt,
override_system_prompt=child_override_prompt,
tool_registry=child_tools,
agent_manager=self.agent_manager,
parent_agent_id=self.managed_agent_id,
@@ -2329,11 +2571,12 @@ class LocalCodingAgent:
summary_lines.append('Final delegated output:')
summary_lines.append(child_result.final_output)
return ToolExecutionResult(
name='delegate_agent',
name=tool_name,
ok=True,
content='\n'.join(summary_lines).strip(),
metadata={
'action': 'delegate_agent',
'action': 'Agent',
'subagent_type': agent_def.agent_type,
'child_session_id': child_result.session_id,
'child_session_ids': child_session_ids,
'child_turns': child_result.turns,
@@ -2562,7 +2805,7 @@ class LocalCodingAgent:
'message_count': len(plugin_delegate_after),
}
)
if tool_call.name != 'delegate_agent':
if tool_call.name not in ('Agent', 'delegate_agent'):
return
delegate_batches = metadata.get('delegate_batches')
if isinstance(delegate_batches, list):
@@ -3037,7 +3280,7 @@ class LocalCodingAgent:
'session_turns': previous_turns + result.turns,
'tool_calls': previous_tool_calls + result.tool_calls,
'delegated_tasks': sum(
1 for entry in result.file_history if entry.get('action') == 'delegate_agent'
1 for entry in result.file_history if entry.get('action') in ('delegate_agent', 'Agent')
),
}
stored = StoredAgentSession(
src/agent_tools.py
+115 -4
View File
@@ -1035,9 +1035,82 @@ def default_tool_registry() -> dict[str, AgentTool]:
},
handler=_todo_write,
),
AgentTool(
name='Agent',
description=(
'Launch a new agent to handle complex, multi-step tasks. '
'Each agent type has specific capabilities and tools available to it.'
),
parameters={
'type': 'object',
'properties': {
'description': {
'type': 'string',
'description': 'A short (3-5 word) description of the task',
},
'prompt': {
'type': 'string',
'description': 'The task for the agent to perform',
},
'subagent_type': {
'type': 'string',
'description': 'The type of specialized agent to use for this task',
},
'model': {
'type': 'string',
'enum': ['sonnet', 'opus', 'haiku'],
'description': 'Optional model override for this agent',
},
'run_in_background': {
'type': 'boolean',
'description': 'Set to true to run this agent in the background',
},
'isolation': {
'type': 'string',
'enum': ['worktree'],
'description': 'Isolation mode. "worktree" creates a temporary git worktree.',
},
'subtasks': {
'type': 'array',
'items': {
'oneOf': [
{'type': 'string'},
{
'type': 'object',
'properties': {
'prompt': {'type': 'string'},
'label': {'type': 'string'},
'max_turns': {'type': 'integer', 'minimum': 1, 'maximum': 20},
'resume_session_id': {'type': 'string'},
'session_id': {'type': 'string'},
'depends_on': {
'type': 'array',
'items': {'type': 'string'},
},
},
'required': ['prompt'],
},
]
},
},
'resume_session_id': {'type': 'string'},
'session_id': {'type': 'string'},
'max_turns': {'type': 'integer', 'minimum': 1, 'maximum': 20},
'allow_write': {'type': 'boolean'},
'allow_shell': {'type': 'boolean'},
'include_parent_context': {'type': 'boolean'},
'continue_on_error': {'type': 'boolean'},
'max_failures': {'type': 'integer', 'minimum': 0, 'maximum': 20},
'strategy': {'type': 'string'},
},
'required': ['description', 'prompt'],
},
handler=_agent_tool_placeholder,
),
# Keep legacy name for backward compatibility
AgentTool(
name='delegate_agent',
description='Delegate a subtask to a nested Python coding agent and return its summary.',
description='(Legacy) Delegate a subtask to a nested agent. Prefer using the Agent tool instead.',
parameters={
'type': 'object',
'properties': {
@@ -1076,7 +1149,29 @@ def default_tool_registry() -> dict[str, AgentTool]:
'strategy': {'type': 'string'},
},
},
handler=_delegate_agent_placeholder,
handler=_agent_tool_placeholder,
),
AgentTool(
name='Skill',
description=(
'Execute a skill within the main conversation. '
'Skills provide specialized capabilities and domain knowledge.'
),
parameters={
'type': 'object',
'properties': {
'skill': {
'type': 'string',
'description': 'The skill name. E.g., "commit", "compact", or "help"',
},
'args': {
'type': 'string',
'description': 'Optional arguments for the skill',
},
},
'required': ['skill'],
},
handler=_execute_skill,
),
]
return {tool.name: tool for tool in tools}
@@ -2754,12 +2849,28 @@ def _stream_bash(
)
def _delegate_agent_placeholder(
def _agent_tool_placeholder(
arguments: dict[str, Any],
context: ToolExecutionContext,
) -> str:
raise ToolExecutionError(
'delegate_agent must be handled by the runtime and is not available as a standalone tool handler'
'Agent/delegate_agent must be handled by the runtime and is not available as a standalone tool handler'
)
def _execute_skill(
arguments: dict[str, Any],
context: ToolExecutionContext,
) -> str | tuple[str, dict[str, Any]]:
"""Execute a skill (slash command) from the Skill tool.
The skill tool must be handled specially by the runtime because it
needs access to the agent to invoke slash commands. This handler
is a placeholder that raises — the runtime intercepts `Skill` tool
calls before they reach this handler.
"""
raise ToolExecutionError(
'Skill must be handled by the runtime and is not available as a standalone tool handler'
)
src/builtin_agents.py
+426
View File
@@ -0,0 +1,426 @@
"""Built-in agent type definitions.
Mirrors the npm ``src/tools/AgentTool/built-in/`` directory. Each agent
type defines its model, tool restrictions, system prompt, and behavioral
flags. The runtime uses these definitions to configure child agents when
the ``Agent`` tool is invoked with a ``subagent_type`` parameter.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
# ---------------------------------------------------------------------------
# Agent definition dataclass
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class AgentDefinition:
"""A single agent type definition (built-in, user, or plugin)."""
agent_type: str
when_to_use: str
system_prompt: str = ''
model: str | None = None # 'sonnet', 'opus', 'haiku', 'inherit', or None (default)
tools: tuple[str, ...] | None = None # Allow-list; None means all
disallowed_tools: tuple[str, ...] = () # Deny-list
color: str | None = None
background: bool = False
one_shot: bool = False
omit_claude_md: bool = False
permission_mode: str | None = None # 'dontAsk', 'plan', etc.
max_turns: int | None = None
critical_system_reminder: str | None = None
source: str = 'built-in'
# ---------------------------------------------------------------------------
# Disallowed tool sets (mirrors npm constants/tools.ts)
# ---------------------------------------------------------------------------
ALL_AGENT_DISALLOWED_TOOLS = frozenset({
'task_output', 'plan_get', 'update_plan', 'plan_clear',
'ask_user_question', 'task_stop',
})
"""Tools disallowed for all child agents by default."""
EXPLORE_PLAN_DISALLOWED_TOOLS = frozenset({
'delegate_agent', 'Agent',
'plan_clear', 'update_plan', 'plan_get',
'edit_file', 'write_file', 'notebook_edit',
})
"""Tools disallowed for read-only agents (Explore, Plan, verification)."""
# ---------------------------------------------------------------------------
# System prompts
# ---------------------------------------------------------------------------
_READ_ONLY_PREAMBLE = """\
CRITICAL: You are in READ-ONLY mode. You are STRICTLY PROHIBITED from:
- Using Edit, Write, or NotebookEdit tools
- Creating, modifying, or deleting any files
- Using Bash for any write operations (no mkdir, touch, rm, cp, mv, \
git add, git commit, npm install, pip install, or any file creation/modification)
- Using redirect operators (>, >>) or heredocs in Bash
- Installing packages or dependencies
You may ONLY use Bash for read-only operations: ls, git status, git log, \
git diff, find, cat, head, tail.
Any attempt to modify files will fail and waste your limited turns."""
_GENERAL_PURPOSE_SYSTEM_PROMPT = """\
You are an agent for Claw Code Python, a Python reimplementation of a \
Claude Code-style coding agent. Given the user's message, you should use \
the tools available to complete the task. Complete the task fully — don't \
gold-plate, but don't leave it half-done.
## Strengths
- Searching code, configs, and patterns across large codebases
- Analyzing multiple files to understand architecture
- Investigating complex questions that need multi-file context
- Multi-step research and implementation tasks
## Guidelines
- Search broadly first when the location of relevant code is unknown
- Use read_file for specific known paths; use glob_search and grep_search \
for discovery
- Start broad, then narrow down to specifics
- Be thorough — check multiple locations and naming conventions
- NEVER create files unless it is absolutely necessary for achieving your goal
- NEVER proactively create documentation files (*.md) or README files \
unless explicitly requested
Your response should be a concise report covering what was done and key \
findings."""
_EXPLORE_SYSTEM_PROMPT = f"""\
{_READ_ONLY_PREAMBLE}
You are a file search specialist. Your role is to rapidly find files, \
search code, and analyze file contents.
## Strengths
- Rapidly finding files using glob patterns
- Searching code with regex patterns
- Reading and analyzing file contents
## Guidelines
- Use glob_search for broad file pattern matching
- Use grep_search for content search with regex
- Use read_file when you know a specific file path
- Use Bash ONLY for read-only operations (ls, git status, git log, \
git diff, find, cat, head, tail)
- Adapt search approach based on the thoroughness level specified in \
the prompt (quick / medium / very thorough)
- Make efficient use of tools — spawn multiple parallel tool calls \
for grepping and reading files when possible
- Communicate your findings as a regular message — do NOT attempt to \
create files
- Complete the search request efficiently and report findings clearly"""
_PLAN_SYSTEM_PROMPT = f"""\
{_READ_ONLY_PREAMBLE}
You are a software architect and planning specialist. Your role is to \
explore the codebase, understand the architecture, and design \
implementation plans.
## Process
1. **Understand Requirements**: Focus on the requirements and the \
assigned perspective in the prompt.
2. **Explore Thoroughly**: Read provided files, find existing patterns, \
understand architecture, identify similar features, trace code paths. \
Use grep_search for patterns; use Bash ONLY for read-only operations.
3. **Design Solution**: Create an approach based on the assigned \
perspective. Consider trade-offs and architectural decisions. Follow \
existing patterns.
4. **Detail the Plan**: Provide a step-by-step strategy. Identify \
dependencies and sequencing. Anticipate challenges.
## Required Output
End your response with:
### Critical Files for Implementation
List the 3-5 most important files that will need to be created or \
modified, with a brief note on the purpose of each change.
REMINDER: You can ONLY explore and plan. You CANNOT write, edit, or \
modify any files. You do NOT have access to file editing tools."""
_VERIFICATION_SYSTEM_PROMPT = f"""\
{_READ_ONLY_PREAMBLE}
You are a verification specialist. Your ONLY job is to verify that \
implementation work is correct.
## Two Failure Patterns to Avoid
1. **Verification avoidance**: Don't just read the code and say it \
looks correct. Run actual commands — builds, tests, linters, type \
checks. Reading is not verifying.
2. **Seduction by the first 80%**: The obvious path often works. The \
bugs hide in edge cases, error paths, concurrent scenarios, and \
boundary conditions. Don't stop after the happy path passes.
## What You MUST Do
- Read CLAUDE.md / README for build and test instructions
- Run the build: does it compile / type-check?
- Run the tests: do they all pass? Are there new/modified tests?
- Run linters / formatters if configured
- Check edge cases and error paths manually
- Look for concurrency issues, boundary values, idempotency problems
- You may create ephemeral test scripts in /tmp via Bash redirection
## What You MUST NOT Do
- Modify any project files (you are read-only)
- Install dependencies
- Run git write operations (add, commit, push, etc.)
- Skip running actual commands in favor of "it looks correct"
## Recognized Rationalizations (Don't Fall For These)
- "The code looks correct so I'll skip running tests"
- "I'll just read the file instead of running the command"
- "The test suite is comprehensive so edge cases are covered"
- "This is a small change so verification isn't needed"
## Required Output Format
For each verification step, report:
- **Command run**: exact command
- **Output observed**: actual output (truncated if long)
- **Result**: PASS / FAIL / SKIP (with reason)
End with exactly one of:
- `VERDICT: PASS` — all checks pass, implementation is correct
- `VERDICT: FAIL` — critical issues found (list them)
- `VERDICT: PARTIAL` — some checks pass, some fail or were skipped"""
_STATUSLINE_SETUP_SYSTEM_PROMPT = """\
You are a status line setup specialist. Your job is to create or update \
the statusLine command in the user's Claude Code settings.
## PS1 Conversion Steps
1. Read shell config files in order: ~/.zshrc, ~/.bashrc, ~/.bash_profile, \
~/.profile
2. Extract the PS1 value
3. Convert PS1 escape sequences to shell commands:
- \\\\u → $(whoami)
- \\\\h → $(hostname -s)
- \\\\H → $(hostname)
- \\\\w → $(pwd)
- \\\\W → $(basename "$(pwd)")
- \\\\$ → $
- \\\\n → \\n
- \\\\t → $(date +%H:%M:%S)
- \\\\d → $(date "+%a %b %d")
- \\\\@ → $(date +%I:%M%p)
4. When using ANSI color codes, use printf — don't remove colors
5. If PS1 would have a trailing "$" or ">", remove them
6. If no PS1 found and no other instructions, ask for further instructions
## Configuration
Update ~/.claude/settings.json (or the symlink target) with the \
statusLine command. Preserve all existing settings when updating.
At the end, inform the parent agent that the "statusline-setup" agent \
must be used for any further status line changes."""
_CLAUDE_CODE_GUIDE_SYSTEM_PROMPT = """\
You are a documentation and feature guidance specialist for Claude Code.
## Your Three Domains
1. **Claude Code (the CLI tool)**: features, hooks, slash commands, MCP \
servers, settings, IDE integrations, keyboard shortcuts
2. **Claude Agent SDK**: building custom agents
3. **Claude API (Anthropic API)**: API usage, tool use, SDK usage
## Approach
1. Determine which domain the question falls into
2. Search for relevant documentation in the codebase
3. Provide guidance with code snippets where helpful
4. Reference specific documentation sections
## Guidelines
- Prioritize official documentation and actual codebase behavior
- Keep responses concise and actionable
- Include code snippets for configuration and usage examples
- If unsure, say so rather than guessing"""
# ---------------------------------------------------------------------------
# Built-in agent instances
# ---------------------------------------------------------------------------
GENERAL_PURPOSE_AGENT = AgentDefinition(
agent_type='general-purpose',
when_to_use=(
'General-purpose agent for researching complex questions, searching '
'for code, and executing multi-step tasks. When you are searching '
'for a keyword or file and are not confident that you will find the '
'right match in the first few tries use this agent to perform the '
'search for you.'
),
system_prompt=_GENERAL_PURPOSE_SYSTEM_PROMPT,
tools=None, # all tools
)
EXPLORE_AGENT = AgentDefinition(
agent_type='Explore',
when_to_use=(
'Fast agent specialized for exploring codebases. Use this when you '
'need to quickly find files by patterns (eg. "src/components/**/*.tsx"), '
'search code for keywords (eg. "API endpoints"), or answer questions '
'about the codebase (eg. "how do API endpoints work?"). When calling '
'this agent, specify the desired thoroughness level: "quick" for basic '
'searches, "medium" for moderate exploration, or "very thorough" for '
'comprehensive analysis across multiple locations and naming conventions.'
),
system_prompt=_EXPLORE_SYSTEM_PROMPT,
model='haiku',
disallowed_tools=tuple(EXPLORE_PLAN_DISALLOWED_TOOLS),
one_shot=True,
omit_claude_md=True,
)
PLAN_AGENT = AgentDefinition(
agent_type='Plan',
when_to_use=(
'Software architect agent for designing implementation plans. Use '
'this when you need to plan the implementation strategy for a task. '
'Returns step-by-step plans, identifies critical files, and considers '
'architectural trade-offs.'
),
system_prompt=_PLAN_SYSTEM_PROMPT,
model='inherit',
disallowed_tools=tuple(EXPLORE_PLAN_DISALLOWED_TOOLS),
one_shot=True,
omit_claude_md=True,
)
VERIFICATION_AGENT = AgentDefinition(
agent_type='verification',
when_to_use=(
'Use this agent to verify that implementation work is correct before '
'reporting completion. Invoke after non-trivial tasks (3+ file edits, '
'backend/API changes, infrastructure changes). Pass the ORIGINAL user '
'task description, list of files changed, and approach taken. The '
'agent runs builds, tests, linters, and checks to produce a '
'PASS/FAIL/PARTIAL verdict with evidence.'
),
system_prompt=_VERIFICATION_SYSTEM_PROMPT,
model='inherit',
color='red',
background=True,
disallowed_tools=tuple(EXPLORE_PLAN_DISALLOWED_TOOLS),
critical_system_reminder=(
'CRITICAL: This is a VERIFICATION-ONLY task. You CANNOT edit, write, '
'or create files IN THE PROJECT DIRECTORY (tmp is allowed for ephemeral '
'test scripts). You MUST end with VERDICT: PASS, VERDICT: FAIL, or '
'VERDICT: PARTIAL.'
),
)
STATUSLINE_SETUP_AGENT = AgentDefinition(
agent_type='statusline-setup',
when_to_use=(
"Use this agent to configure the user's Claude Code status line setting."
),
system_prompt=_STATUSLINE_SETUP_SYSTEM_PROMPT,
model='sonnet',
color='orange',
tools=('read_file', 'edit_file'),
)
CLAUDE_CODE_GUIDE_AGENT = AgentDefinition(
agent_type='claude-code-guide',
when_to_use=(
'Use this agent when the user asks questions ("Can Claude...", '
'"Does Claude...", "How do I...") about: (1) Claude Code (the CLI '
'tool) - features, hooks, slash commands, MCP servers, settings, IDE '
'integrations, keyboard shortcuts; (2) Claude Agent SDK - building '
'custom agents; (3) Claude API (formerly Anthropic API) - API usage, '
'tool use, Anthropic SDK usage. **IMPORTANT:** Before spawning a new '
'agent, check if there is already a running or recently completed '
'claude-code-guide agent that you can continue via SendMessage.'
),
system_prompt=_CLAUDE_CODE_GUIDE_SYSTEM_PROMPT,
model='haiku',
permission_mode='dontAsk',
tools=('glob_search', 'grep_search', 'read_file', 'web_fetch', 'web_search'),
)
# ---------------------------------------------------------------------------
# Agent registry
# ---------------------------------------------------------------------------
ONE_SHOT_AGENT_TYPES = frozenset({'Explore', 'Plan'})
"""Agent types that run once and return a report (no agentId / SendMessage)."""
_BUILTIN_AGENTS: tuple[AgentDefinition, ...] = (
GENERAL_PURPOSE_AGENT,
EXPLORE_AGENT,
PLAN_AGENT,
VERIFICATION_AGENT,
STATUSLINE_SETUP_AGENT,
CLAUDE_CODE_GUIDE_AGENT,
)
def get_builtin_agents() -> tuple[AgentDefinition, ...]:
"""Return all built-in agent definitions."""
return _BUILTIN_AGENTS
def get_agent_definition(agent_type: str) -> AgentDefinition | None:
"""Look up a built-in agent definition by type name (case-sensitive)."""
for agent in _BUILTIN_AGENTS:
if agent.agent_type == agent_type:
return agent
return None
def get_agent_types() -> list[str]:
"""Return sorted list of all available agent type names."""
return sorted(agent.agent_type for agent in _BUILTIN_AGENTS)
def format_agent_listing(agents: tuple[AgentDefinition, ...] | None = None) -> str:
"""Format agent types for inclusion in the Agent tool prompt.
Mirrors the npm ``formatAgentLine`` helper.
"""
if agents is None:
agents = _BUILTIN_AGENTS
lines: list[str] = []
for agent in agents:
tools_desc = _describe_tools(agent)
lines.append(f'- {agent.agent_type}: {agent.when_to_use} (Tools: {tools_desc})')
return '\n'.join(lines)
def _describe_tools(agent: AgentDefinition) -> str:
"""Describe the tool access for an agent definition."""
if agent.tools is not None:
return ', '.join(agent.tools) if agent.tools else 'none'
if agent.disallowed_tools:
denied = ', '.join(sorted(agent.disallowed_tools))
return f'All tools except {denied}'
return '*'
src/compact.py
+212 -31
View File
@@ -9,6 +9,9 @@ Mirrors the npm ``src/services/compact/compact.ts`` and
(``get_compact_user_summary_message``).
- The core ``compact_conversation`` entry point that an
``/compact`` slash command or auto-compact subsystem can call.
- PTL retry loop: drops oldest API-round groups when the compact
request itself hits prompt-too-long (up to ``MAX_PTL_RETRIES``).
- Circuit-breaker tracking for consecutive failures.
"""
from __future__ import annotations
@@ -38,11 +41,21 @@ ERROR_INCOMPLETE_RESPONSE = (
'The conversation was not compacted.'
)
ERROR_USER_ABORT = 'Compaction canceled.'
ERROR_PROMPT_TOO_LONG = (
'The compact request itself was too long even after retry truncation.'
)
MAX_COMPACT_FAILURES = 3
"""Circuit-breaker – stop retrying auto-compact after this many consecutive
failures (mirrors the npm implementation)."""
MAX_PTL_RETRIES = 3
"""Maximum number of prompt-too-long retry attempts during compaction."""
PTL_RETRY_MARKER = '[compact_ptl_retry_marker]'
"""Synthetic user message prepended when dropping the first API-round group
leaves an assistant message at position 0."""
# ---------------------------------------------------------------------------
# Prompt construction (npm ``src/services/compact/prompt.ts``)
# ---------------------------------------------------------------------------
@@ -272,6 +285,103 @@ def get_compact_user_summary_message(
return base
# ---------------------------------------------------------------------------
# API-round grouping (mirrors npm groupMessagesByApiRound)
# ---------------------------------------------------------------------------
def group_messages_by_api_round(
messages: list[AgentMessage],
) -> list[list[AgentMessage]]:
"""Group messages at API-round boundaries.
A new group starts when an assistant message with a different
``message_id`` than the previous assistant message is encountered.
"""
groups: list[list[AgentMessage]] = []
current: list[AgentMessage] = []
last_assistant_id: str | None = None
for msg in messages:
if (
msg.role == 'assistant'
and msg.message_id != last_assistant_id
and current
):
groups.append(current)
current = [msg]
else:
current.append(msg)
if msg.role == 'assistant':
last_assistant_id = msg.message_id
if current:
groups.append(current)
return groups
def truncate_head_for_ptl_retry(
messages: list[AgentMessage],
model: str = '',
token_gap: int | None = None,
) -> list[AgentMessage] | None:
"""Drop oldest API-round groups to free space for the compact request.
If *token_gap* is provided, drops enough groups to cover that many
tokens. Otherwise falls back to dropping ~20% of groups.
Returns ``None`` if nothing can be dropped without emptying the list.
"""
# Strip a previous PTL_RETRY_MARKER to prevent stalling
working = list(messages)
if (
working
and working[0].role == 'user'
and working[0].content == PTL_RETRY_MARKER
):
working = working[1:]
groups = group_messages_by_api_round(working)
if len(groups) < 2:
return None
if token_gap is not None and token_gap > 0:
accumulated = 0
drop_count = 0
for group in groups:
group_tokens = sum(estimate_tokens(m.content, model) for m in group)
accumulated += group_tokens
drop_count += 1
if accumulated >= token_gap:
break
else:
# Fallback: drop ~20% of groups
drop_count = max(1, len(groups) // 5)
drop_count = min(drop_count, len(groups) - 1)
if drop_count < 1:
return None
remaining: list[AgentMessage] = []
for group in groups[drop_count:]:
remaining.extend(group)
if not remaining:
return None
# If the first remaining message is an assistant message, prepend a
# synthetic user marker so the API contract is satisfied.
if remaining[0].role == 'assistant':
marker = AgentMessage(
role='user',
content=PTL_RETRY_MARKER,
message_id='ptl_retry_marker',
metadata={'kind': 'ptl_retry_marker', 'is_meta': True},
)
remaining = [marker] + remaining
return remaining
# ---------------------------------------------------------------------------
# Compaction result
# ---------------------------------------------------------------------------
@@ -285,15 +395,54 @@ class CompactionResult:
messages_to_keep: list[AgentMessage] = field(default_factory=list)
pre_compact_token_count: int = 0
post_compact_token_count: int = 0
true_post_compact_token_count: int = 0
summary_text: str = ''
usage: UsageStats = field(default_factory=UsageStats)
error: str | None = None
ptl_retries: int = 0
# ---------------------------------------------------------------------------
# Core compaction logic
# ---------------------------------------------------------------------------
def _is_prompt_too_long_response(content: str) -> bool:
"""Check if a model response indicates a prompt-too-long error.
Some models embed the error in the response text rather than raising.
"""
lower = content.lower()
return (
'prompt is too long' in lower
or 'prompt_too_long' in lower
or 'context_length_exceeded' in lower
)
def _call_compact_model(
agent: 'LocalCodingAgent',
api_messages: list[dict[str, Any]],
) -> tuple[str | None, 'UsageStats', str | None]:
"""Call the model for compaction, returning (content, usage, error).
Returns (None, usage, error_string) on failure.
"""
try:
turn = agent.client.complete(api_messages, tools=[])
except Exception as exc:
error_str = str(exc)
if 'prompt' in error_str.lower() and 'long' in error_str.lower():
return None, UsageStats(), 'prompt_too_long'
return None, UsageStats(), error_str
raw = turn.content or ''
if not raw.strip():
return None, turn.usage, 'empty_response'
if _is_prompt_too_long_response(raw):
return None, turn.usage, 'prompt_too_long'
return raw, turn.usage, None
def compact_conversation(
agent: 'LocalCodingAgent',
custom_instructions: str | None = None,
@@ -303,8 +452,10 @@ def compact_conversation(
1. Build the compact prompt (9-section template).
2. Collect the session messages to summarise.
3. Send them + the compact prompt to the model.
4. Parse ``<summary>`` from the response.
5. Replace session messages with:
4. On prompt-too-long, retry by dropping oldest API-round groups
(up to ``MAX_PTL_RETRIES`` attempts).
5. Parse ``<summary>`` from the response.
6. Replace session messages with:
boundary marker → summary user message → preserved tail.
Returns a :class:`CompactionResult` with diagnostics.
@@ -317,13 +468,10 @@ def compact_conversation(
)
# ---- Determine which messages to compact vs preserve ----
# We keep the most recent ``preserve_count`` messages untouched.
preserve_count = max(
getattr(agent.runtime_config, 'compact_preserve_messages', 4), 1
)
# Identify the prefix count (system-injected messages that precede the
# real conversation, e.g. a compaction-replay boundary).
prefix_count = 0
for msg in session.messages:
if msg.metadata.get('kind') == 'compact_boundary':
@@ -341,7 +489,7 @@ def compact_conversation(
error=ERROR_NOT_ENOUGH_MESSAGES,
)
candidates = session.messages[prefix_count:compact_end]
candidates = list(session.messages[prefix_count:compact_end])
preserved_tail = list(session.messages[compact_end:])
if not candidates:
@@ -354,39 +502,67 @@ def compact_conversation(
model = agent.model_config.model
pre_tokens = sum(estimate_tokens(m.content, model) for m in session.messages)
# ---- Build the compact request messages ----
# ---- Build the compact request ----
compact_prompt = get_compact_prompt(custom_instructions)
# We send the system prompt + candidate messages + the compact prompt as
# a user message. The model returns the summary.
api_messages: list[dict[str, Any]] = []
# ---- PTL retry loop ----
messages_to_summarize = candidates
ptl_retries = 0
total_usage = UsageStats()
raw_summary: str | None = None
# System prompt (from session)
for part in session.system_prompt_parts:
if part.strip():
api_messages.append({'role': 'system', 'content': part})
for attempt in range(MAX_PTL_RETRIES + 1):
api_messages: list[dict[str, Any]] = []
# Candidate messages (the ones to be summarised)
for msg in candidates:
api_messages.append(msg.to_openai_message())
for part in session.system_prompt_parts:
if part.strip():
api_messages.append({'role': 'system', 'content': part})
# The compact prompt as the final user turn
api_messages.append({'role': 'user', 'content': compact_prompt})
for msg in messages_to_summarize:
api_messages.append(msg.to_openai_message())
# ---- Call the model ----
try:
turn = agent.client.complete(api_messages, tools=[])
except Exception as exc:
return CompactionResult(
boundary_message=_build_boundary(f'Compact API call failed: {exc}'),
error=str(exc),
api_messages.append({'role': 'user', 'content': compact_prompt})
content, usage, error = _call_compact_model(agent, api_messages)
total_usage = total_usage + usage
if error != 'prompt_too_long':
raw_summary = content
break
# PTL error — try truncating oldest API-round groups
ptl_retries += 1
if attempt >= MAX_PTL_RETRIES:
return CompactionResult(
boundary_message=_build_boundary(
f'Compact request was too long after {ptl_retries} retries.'
),
error=ERROR_PROMPT_TOO_LONG,
usage=total_usage,
ptl_retries=ptl_retries,
)
truncated = truncate_head_for_ptl_retry(
messages_to_summarize, model=model,
)
if truncated is None:
return CompactionResult(
boundary_message=_build_boundary(
'Cannot truncate further for compact retry.'
),
error=ERROR_PROMPT_TOO_LONG,
usage=total_usage,
ptl_retries=ptl_retries,
)
messages_to_summarize = truncated
raw_summary = turn.content or ''
if not raw_summary.strip():
if raw_summary is None:
error_msg = error or ERROR_INCOMPLETE_RESPONSE
return CompactionResult(
boundary_message=_build_boundary('Model returned empty summary.'),
error=ERROR_INCOMPLETE_RESPONSE,
boundary_message=_build_boundary(f'Compaction failed: {error_msg}'),
error=error_msg,
usage=total_usage,
ptl_retries=ptl_retries,
)
# ---- Format the summary ----
@@ -422,8 +598,13 @@ def compact_conversation(
messages_to_keep=preserved_tail,
pre_compact_token_count=pre_tokens,
post_compact_token_count=post_tokens,
true_post_compact_token_count=sum(
estimate_tokens(m.content, model)
for m in [boundary, summary_msg] + preserved_tail
),
summary_text=summary_text,
usage=turn.usage,
usage=total_usage,
ptl_retries=ptl_retries,
)
src/microcompact.py
+236
View File
@@ -0,0 +1,236 @@
"""Microcompact service — lightweight tool-result clearing for context efficiency.
Mirrors the npm ``src/services/compact/microCompact.ts`` module.
Provides time-based microcompaction: when a significant gap exists since the
last assistant message (indicating cache has expired), old tool results are
replaced with a short cleared marker to reduce context size on the next
API call.
"""
from __future__ import annotations
import time
from dataclasses import dataclass, field
from typing import Any
from .agent_context_usage import estimate_tokens
from .agent_session import AgentMessage
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
TIME_BASED_MC_CLEARED_MESSAGE = '[Old tool result content cleared]'
"""Replacement content for cleared tool results."""
IMAGE_MAX_TOKEN_SIZE = 2000
"""Fixed token estimate for image/document blocks."""
# Tools whose results can be safely cleared during microcompaction.
COMPACTABLE_TOOLS: frozenset[str] = frozenset({
'read_file',
'bash',
'grep_search',
'glob_search',
'web_search',
'web_fetch',
'edit_file',
'write_file',
})
DEFAULT_GAP_THRESHOLD_MINUTES = 60.0
"""Minimum gap (in minutes) since the last assistant message before
time-based microcompact fires. Mirrors the npm default."""
DEFAULT_KEEP_RECENT = 3
"""Number of most-recent compactable tool results to always preserve."""
# ---------------------------------------------------------------------------
# Result type
# ---------------------------------------------------------------------------
@dataclass
class MicrocompactResult:
"""Outcome of a microcompact pass."""
messages: list[AgentMessage]
cleared_tool_count: int = 0
kept_tool_count: int = 0
estimated_tokens_saved: int = 0
triggered: bool = False
gap_minutes: float = 0.0
# ---------------------------------------------------------------------------
# Time-based microcompact
# ---------------------------------------------------------------------------
def _find_last_assistant_timestamp(messages: list[AgentMessage]) -> float | None:
"""Find the timestamp of the last assistant message.
Returns seconds since epoch, or ``None`` if no assistant message has a
``timestamp`` metadata entry.
"""
for msg in reversed(messages):
if msg.role != 'assistant':
continue
ts = msg.metadata.get('timestamp')
if isinstance(ts, (int, float)):
return float(ts)
if isinstance(ts, str):
try:
from datetime import datetime, timezone
dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
return dt.timestamp()
except (ValueError, TypeError):
pass
# Fall back to message creation time tracked by the session
created = msg.metadata.get('created_at')
if isinstance(created, (int, float)):
return float(created)
return None
def _collect_compactable_tool_ids(
messages: list[AgentMessage],
) -> list[str]:
"""Collect tool_call_ids for tool results that can be safely cleared.
Walks messages in order and returns IDs for tool-result messages whose
name is in :data:`COMPACTABLE_TOOLS`.
"""
ids: list[str] = []
for msg in messages:
if msg.role != 'tool':
continue
if not msg.tool_call_id:
continue
tool_name = msg.name or msg.metadata.get('tool_name', '')
if tool_name in COMPACTABLE_TOOLS:
ids.append(msg.tool_call_id)
return ids
def evaluate_time_based_trigger(
messages: list[AgentMessage],
*,
gap_threshold_minutes: float = DEFAULT_GAP_THRESHOLD_MINUTES,
) -> float | None:
"""Return the gap in minutes since the last assistant message, or ``None``.
Returns ``None`` when the trigger does not fire (gap below threshold,
no assistant messages, etc.).
"""
ts = _find_last_assistant_timestamp(messages)
if ts is None:
return None
gap_seconds = time.time() - ts
if gap_seconds < 0:
return None
gap_minutes = gap_seconds / 60.0
if gap_minutes < gap_threshold_minutes:
return None
return gap_minutes
def microcompact_messages(
messages: list[AgentMessage],
*,
model: str = '',
gap_threshold_minutes: float = DEFAULT_GAP_THRESHOLD_MINUTES,
keep_recent: int = DEFAULT_KEEP_RECENT,
) -> MicrocompactResult:
"""Run time-based microcompaction on session messages.
When the gap since the last assistant message exceeds
*gap_threshold_minutes*, old tool results (beyond the *keep_recent*
most recent) are replaced with a short cleared marker.
This is useful when the server-side prompt cache has expired and the
entire prefix will be rewritten anyway — clearing old tool results
shrinks the rewrite payload.
Parameters
----------
messages:
The session messages to process (not mutated — new list returned).
model:
Model name for token estimation.
gap_threshold_minutes:
Minimum idle gap before trigger fires.
keep_recent:
Number of most-recent compactable tool results to keep.
Returns
-------
MicrocompactResult
A result containing the (possibly modified) message list and
diagnostic counters.
"""
gap_minutes = evaluate_time_based_trigger(
messages, gap_threshold_minutes=gap_threshold_minutes,
)
if gap_minutes is None:
return MicrocompactResult(messages=messages)
# Collect compactable tool IDs in order
compactable_ids = _collect_compactable_tool_ids(messages)
if not compactable_ids:
return MicrocompactResult(messages=messages, gap_minutes=gap_minutes)
# Keep the most recent `keep_recent` tools untouched
keep_count = max(1, keep_recent)
if len(compactable_ids) <= keep_count:
return MicrocompactResult(
messages=messages,
kept_tool_count=len(compactable_ids),
gap_minutes=gap_minutes,
)
clear_ids = set(compactable_ids[:-keep_count])
keep_ids = set(compactable_ids[-keep_count:])
# Build a new message list with cleared tool results
new_messages: list[AgentMessage] = []
tokens_saved = 0
cleared_count = 0
for msg in messages:
if (
msg.role == 'tool'
and msg.tool_call_id
and msg.tool_call_id in clear_ids
):
original_tokens = estimate_tokens(msg.content, model)
replacement_tokens = estimate_tokens(TIME_BASED_MC_CLEARED_MESSAGE, model)
tokens_saved += max(original_tokens - replacement_tokens, 0)
cleared_count += 1
# Create a new message with cleared content
new_msg = AgentMessage(
role=msg.role,
content=TIME_BASED_MC_CLEARED_MESSAGE,
name=msg.name,
tool_call_id=msg.tool_call_id,
message_id=msg.message_id,
metadata={
**msg.metadata,
'microcompact_cleared': True,
'original_token_estimate': original_tokens,
},
)
new_messages.append(new_msg)
else:
new_messages.append(msg)
return MicrocompactResult(
messages=new_messages,
cleared_tool_count=cleared_count,
kept_tool_count=len(keep_ids),
estimated_tokens_saved=tokens_saved,
triggered=cleared_count > 0,
gap_minutes=gap_minutes,
)