diff --git a/.gitignore b/.gitignore
index f786461..ed4e265 100644
--- a/.gitignore
+++ b/.gitignore
@@ -34,3 +34,4 @@ test_cases
e-commerce
benchmarks/data/*.jsonl
benchmarks/data/manifest.json
+claude-code-sourcemap-main
\ No newline at end of file
diff --git a/CLAUDE.md b/CLAUDE.md
new file mode 100644
index 0000000..ca1043e
--- /dev/null
+++ b/CLAUDE.md
@@ -0,0 +1,65 @@
+# CLAUDE.md
+
+Behavioral guidelines to reduce common LLM coding mistakes. Merge with project-specific instructions as needed.
+
+**Tradeoff:** These guidelines bias toward caution over speed. For trivial tasks, use judgment.
+
+## 1. Think Before Coding
+
+**Don't assume. Don't hide confusion. Surface tradeoffs.**
+
+Before implementing:
+- State your assumptions explicitly. If uncertain, ask.
+- If multiple interpretations exist, present them - don't pick silently.
+- If a simpler approach exists, say so. Push back when warranted.
+- If something is unclear, stop. Name what's confusing. Ask.
+
+## 2. Simplicity First
+
+**Minimum code that solves the problem. Nothing speculative.**
+
+- No features beyond what was asked.
+- No abstractions for single-use code.
+- No "flexibility" or "configurability" that wasn't requested.
+- No error handling for impossible scenarios.
+- If you write 200 lines and it could be 50, rewrite it.
+
+Ask yourself: "Would a senior engineer say this is overcomplicated?" If yes, simplify.
+
+## 3. Surgical Changes
+
+**Touch only what you must. Clean up only your own mess.**
+
+When editing existing code:
+- Don't "improve" adjacent code, comments, or formatting.
+- Don't refactor things that aren't broken.
+- Match existing style, even if you'd do it differently.
+- If you notice unrelated dead code, mention it - don't delete it.
+
+When your changes create orphans:
+- Remove imports/variables/functions that YOUR changes made unused.
+- Don't remove pre-existing dead code unless asked.
+
+The test: Every changed line should trace directly to the user's request.
+
+## 4. Goal-Driven Execution
+
+**Define success criteria. Loop until verified.**
+
+Transform tasks into verifiable goals:
+- "Add validation" → "Write tests for invalid inputs, then make them pass"
+- "Fix the bug" → "Write a test that reproduces it, then make it pass"
+- "Refactor X" → "Ensure tests pass before and after"
+
+For multi-step tasks, state a brief plan:
+```
+1. [Step] → verify: [check]
+2. [Step] → verify: [check]
+3. [Step] → verify: [check]
+```
+
+Strong success criteria let you loop independently. Weak criteria ("make it work") require constant clarification.
+
+---
+
+**These guidelines are working if:** fewer unnecessary changes in diffs, fewer rewrites due to overcomplication, and clarifying questions come before implementation rather than after mistakes.
\ No newline at end of file
diff --git a/PARITY_CHECKLIST.md b/PARITY_CHECKLIST.md
index 9707b66..bcf93cd 100644
--- a/PARITY_CHECKLIST.md
+++ b/PARITY_CHECKLIST.md
@@ -363,7 +363,7 @@ Missing npm slash commands (from `src/commands/` — 80+ commands total):
## 6. Built-in Tools
-### Tools implemented in Python (58 tools):
+### Tools implemented in Python (61 tools):
- [x] `list_dir`
- [x] `read_file`
@@ -428,8 +428,8 @@ Missing npm slash commands (from `src/commands/` — 80+ commands total):
### Tools in npm `tools.ts` not yet ported with full fidelity (40 tool dirs):
Core tools needing full port:
-- [ ] `AgentTool` — Sub-agent spawning with built-in agents (explore, general-purpose, verification, plan, claudeCodeGuide, statusline), fork support, agent memory/snapshots, resume agent, color management
-- [ ] `SkillTool` — Skill execution with bundled skills
+- [x] `AgentTool` — Sub-agent spawning with built-in agents (explore, general-purpose, verification, plan, claudeCodeGuide, statusline) → `src/builtin_agents.py` — remaining: fork support, agent memory/snapshots, resume agent, color management
+- [x] `SkillTool` — Skill execution via slash commands → `src/agent_tools.py`, `src/agent_runtime.py` — remaining: bundled skill definitions, forked skill execution
- [ ] `BriefTool` — Brief mode with attachments and file upload
- [ ] `LSPTool` — Full upstream LSP fidelity beyond the current local heuristic runtime (server-backed diagnostics, go-to-definition, references, hover, symbol search, formatting)
- [ ] `PowerShellTool` — Full PowerShell execution with security, path validation, CLM types, git safety
@@ -737,8 +737,8 @@ Missing:
Working Python runtime today (21,193 lines across 51 source files, 10,480 lines across 37 test files):
- [x] `src/main.py` (1,353 lines)
-- [x] `src/agent_runtime.py` (3,664 lines)
-- [x] `src/agent_tools.py` (2,994 lines)
+- [x] `src/agent_runtime.py` (4,318 lines)
+- [x] `src/agent_tools.py` (3,183 lines)
- [x] `src/agent_prompting.py` (390 lines)
- [x] `src/agent_context.py` (459 lines)
- [x] `src/agent_context_usage.py` (356 lines)
@@ -768,6 +768,8 @@ Working Python runtime today (21,193 lines across 51 source files, 10,480 lines
- [x] `src/tokenizer_runtime.py` (202 lines)
- [x] `src/workflow_runtime.py` (319 lines)
- [x] `src/worktree_runtime.py` (448 lines)
+- [x] `src/builtin_agents.py` (426 lines)
+- [x] `src/microcompact.py` (236 lines)
- [x] Plus 19 supporting modules
Mirrored / scaffold areas needing real implementation:
@@ -785,13 +787,15 @@ Mirrored / scaffold areas needing real implementation:
### Tier 1 — Core Feature Gaps (highest user impact)
- [x] Full BashTool security parity (sed validation, path validation, sandbox, destructive command warnings, command semantics) → `src/bash_security.py`
- [x] LSP tool integration for code intelligence
-- [ ] Full AgentTool with built-in agent types (explore, general-purpose, verification, plan)
+- [x] Full AgentTool with built-in agent types (explore, general-purpose, verification, plan, statusline-setup, claude-code-guide) → `src/builtin_agents.py`, `src/agent_tools.py`, `src/agent_runtime.py`
- [ ] Auto-compact and context collapse from `query.ts`
-- [ ] Full compact service (autoCompact, microCompact, sessionMemoryCompact)
+- [x] Microcompact service (time-based tool-result clearing) → `src/microcompact.py`
+- [x] Compact PTL retry loop and circuit-breaker tracking → `src/compact.py`
+- [ ] Full compact service (sessionMemoryCompact, cached microcompact)
- [ ] Interactive REPL improvements
### Tier 2 — Important Feature Gaps
-- [ ] SkillTool with bundled skills
+- [x] SkillTool for slash command execution → `src/agent_tools.py`, `src/agent_runtime.py`
- [ ] Full MCP service parity (auth, permissions, config, registry)
- [ ] Plugin discovery, loading, and installation
- [ ] Real remote session management (WebSocket, bridge)
diff --git a/src/agent_runtime.py b/src/agent_runtime.py
index 8a5a383..8e550d3 100644
--- a/src/agent_runtime.py
+++ b/src/agent_runtime.py
@@ -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\n{agent_def.critical_system_reminder}\n'
+ 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(
diff --git a/src/agent_tools.py b/src/agent_tools.py
index 317edd5..484ee08 100644
--- a/src/agent_tools.py
+++ b/src/agent_tools.py
@@ -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'
)
diff --git a/src/builtin_agents.py b/src/builtin_agents.py
new file mode 100644
index 0000000..7283fc1
--- /dev/null
+++ b/src/builtin_agents.py
@@ -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 '*'
diff --git a/src/compact.py b/src/compact.py
index 4a322a1..2ed9651 100644
--- a/src/compact.py
+++ b/src/compact.py
@@ -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 ```` 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 ```` 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,
)
diff --git a/src/microcompact.py b/src/microcompact.py
new file mode 100644
index 0000000..71b40ba
--- /dev/null
+++ b/src/microcompact.py
@@ -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,
+ )