diff --git a/PARITY_CHECKLIST.md b/PARITY_CHECKLIST.md index 30ace9a..fd14493 100644 --- a/PARITY_CHECKLIST.md +++ b/PARITY_CHECKLIST.md @@ -26,6 +26,7 @@ Done: - [x] Incremental tool-result streaming for read-only text tools - [x] Mutable tool transcript updates during tool execution - [x] Transcript mutation history for replaced/tombstoned messages +- [x] Assistant streaming and tool-call transcript mutation history - [x] Structured transcript block export for messages, tool calls, and tool results - [x] Resume-time file-history replay reminders - [x] Resume-time file-history snapshot previews for file edits @@ -39,26 +40,42 @@ Done: - [x] Sequential multi-subtask delegation with parent-context carryover - [x] Basic agent-manager lineage tracking for nested agents - [x] Managed agent-group membership tracking with child indices +- [x] Delegated child-session resume by saved session id +- [x] Agent-manager tracking for resumed child-session lineage - [x] Plugin-cache discovery and prompt-context injection - [x] Manifest-based plugin runtime discovery - [x] Manifest-defined plugin hooks for before-prompt and after-turn runtime injection - [x] Manifest-defined plugin tool aliases over base runtime tools +- [x] Manifest-defined executable virtual tools - [x] Manifest-defined plugin tool blocking +- [x] Manifest-defined plugin `beforeTool` guidance - [x] Manifest-defined plugin tool-result guidance injected back into the transcript - [x] Compaction metadata with compacted message ids +- [x] Compaction metadata with preserved-tail ids and compaction depth +- [x] Compaction metadata with compacted/preserved lineage ids and revision summaries +- [x] Snipped-message metadata with source role/kind lineage +- [x] Snipped-message metadata with source lineage id and revision - [x] Resume-time compaction / snipping replay reminder - [x] Query-engine facade that can drive the real Python runtime agent - [x] Query-engine runtime event counters and transcript-kind summaries +- [x] Query-engine runtime mutation counters +- [x] Query-engine stream-level runtime summary event +- [x] Query-engine transcript-store compaction summaries +- [x] Delegate-group and delegated-subtask runtime events +- [x] Query-engine runtime orchestration summaries for group status and child stop reasons +- [x] Query-engine runtime context-reduction summaries +- [x] Query-engine runtime lineage summaries +- [x] Query-engine runtime resumed-child orchestration summaries Missing: - [ ] Full partial tool-result streaming parity across the complete tool surface - [ ] Full rich transcript mutation behavior like the npm runtime - [ ] Full reasoning budgets and task budgets parity -- [ ] Full multi-agent orchestration parity +- [ ] Full multi-agent orchestration parity beyond sequential grouped delegation and resumed-child flows - [ ] Full file history snapshots and replay flows -- [ ] Full executable plugin lifecycle beyond runtime guidance, blocking, and aliases -- [ ] Full session compaction / snipping parity +- [ ] Full executable plugin lifecycle beyond manifest-driven guidance, blocking, aliases, and virtual tools +- [ ] Full session compaction / snipping parity beyond lineage-aware summaries and replay reminders - [ ] Full `QueryEngine.ts` parity ## 2. CLI Entrypoints And Runtime Modes diff --git a/src/agent_manager.py b/src/agent_manager.py index 6643742..a6e4910 100644 --- a/src/agent_manager.py +++ b/src/agent_manager.py @@ -11,6 +11,7 @@ class ManagedAgentRecord: group_id: str | None = None child_index: int | None = None label: str | None = None + resumed_from_session_id: str | None = None session_id: str | None = None session_path: str | None = None status: str = 'running' @@ -45,6 +46,7 @@ class AgentManager: group_id: str | None = None, child_index: int | None = None, label: str | None = None, + resumed_from_session_id: str | None = None, ) -> str: self._counter += 1 agent_id = f'agent_{self._counter}' @@ -55,6 +57,7 @@ class AgentManager: group_id=group_id, child_index=child_index, label=label, + resumed_from_session_id=resumed_from_session_id, ) if group_id is not None: self.register_group_child(group_id, agent_id, child_index=child_index) @@ -110,6 +113,7 @@ class AgentManager: group_id=group_id, child_index=child_index, label=record.label, + resumed_from_session_id=record.resumed_from_session_id, session_id=record.session_id, session_path=record.session_path, status=record.status, @@ -159,6 +163,7 @@ class AgentManager: group_id=record.group_id, child_index=record.child_index, label=record.label, + resumed_from_session_id=record.resumed_from_session_id, session_id=session_id, session_path=session_path, status='completed', @@ -174,6 +179,44 @@ class AgentManager: if record.parent_agent_id == agent_id ) + def group_children(self, group_id: str) -> tuple[ManagedAgentRecord, ...]: + return tuple( + sorted( + ( + record for record in self.records.values() + if record.group_id == group_id + ), + key=lambda record: ( + record.child_index is None, + record.child_index or 0, + record.agent_id, + ), + ) + ) + + def group_summary(self, group_id: str) -> dict[str, object] | None: + group = self.groups.get(group_id) + if group is None: + return None + children = self.group_children(group_id) + stop_reason_counts: dict[str, int] = {} + resumed_children = 0 + for child in children: + if child.resumed_from_session_id: + resumed_children += 1 + stop_reason = child.stop_reason or 'n/a' + stop_reason_counts[stop_reason] = stop_reason_counts.get(stop_reason, 0) + 1 + return { + 'group_id': group.group_id, + 'label': group.label, + 'status': group.status, + 'child_count': len(children), + 'completed_children': group.completed_children, + 'failed_children': group.failed_children, + 'resumed_children': resumed_children, + 'stop_reason_counts': stop_reason_counts, + } + def completed_records(self) -> tuple[ManagedAgentRecord, ...]: return tuple( record for record in self.records.values() if record.status == 'completed' @@ -186,6 +229,10 @@ class AgentManager: ] child_count = sum(1 for record in self.records.values() if record.parent_agent_id) lines.append(f'- Child agents: {child_count}') + resumed_count = sum( + 1 for record in self.records.values() if record.resumed_from_session_id + ) + lines.append(f'- Resumed agents: {resumed_count}') lines.append(f'- Agent groups: {len(self.groups)}') completed_groups = sum(1 for group in self.groups.values() if group.status == 'completed') lines.append(f'- Completed groups: {completed_groups}') @@ -196,6 +243,8 @@ class AgentManager: group_bits.append(f'group={record.group_id}') if record.child_index is not None: group_bits.append(f'child_index={record.child_index}') + if record.resumed_from_session_id is not None: + group_bits.append(f'resumed_from={record.resumed_from_session_id}') group_suffix = f" {' '.join(group_bits)}" if group_bits else '' lines.append( f'- {label}: status={record.status} turns={record.turns} ' @@ -205,9 +254,19 @@ class AgentManager: lines.append(f'- ... plus {len(self.records) - 8} more managed agents') for group in sorted(self.groups.values(), key=lambda item: item.group_id)[:6]: label = group.label or group.group_id + summary = self.group_summary(group.group_id) + if summary is None: + continue + stop_bits = summary['stop_reason_counts'] + stop_suffix = '' + if isinstance(stop_bits, dict) and stop_bits: + stop_suffix = ' stop_reasons=' + ','.join( + f'{name}:{count}' for name, count in sorted(stop_bits.items()) + ) lines.append( f'- {label}: group_status={group.status} children={len(group.child_agent_ids)} ' - f'completed={group.completed_children} failed={group.failed_children}' + f'completed={group.completed_children} failed={group.failed_children} ' + f"resumed={summary['resumed_children']}{stop_suffix}" ) if len(self.groups) > 6: lines.append(f'- ... plus {len(self.groups) - 6} more agent groups') diff --git a/src/agent_runtime.py b/src/agent_runtime.py index 0485f2e..6256db4 100644 --- a/src/agent_runtime.py +++ b/src/agent_runtime.py @@ -4,6 +4,7 @@ from dataclasses import dataclass, field, replace from datetime import datetime, timezone import json from pathlib import Path +from typing import Any from uuid import uuid4 from .agent_manager import AgentManager @@ -40,6 +41,7 @@ from .openai_compat import OpenAICompatClient, OpenAICompatError from .plugin_runtime import PluginRuntime from .session_store import ( StoredAgentSession, + load_agent_session, save_agent_session, serialize_model_config, serialize_runtime_config, @@ -64,12 +66,14 @@ class LocalCodingAgent: parent_agent_id: str | None = None managed_group_id: str | None = None managed_child_index: int | None = None + managed_label: str | None = None plugin_runtime: PluginRuntime | None = None last_session: AgentSessionState | None = field(default=None, init=False, repr=False) last_run_result: AgentRunResult | None = field(default=None, 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) + resume_source_session_id: str | None = field(default=None, init=False, repr=False) def __post_init__(self) -> None: if self.tool_registry is None: @@ -81,9 +85,14 @@ class LocalCodingAgent: self.runtime_config.cwd, tuple(str(path) for path in self.runtime_config.additional_working_directories), ) - plugin_tools = self.plugin_runtime.register_tool_aliases(self.tool_registry) + registry = dict(self.tool_registry) + plugin_tools = self.plugin_runtime.register_tool_aliases(registry) if plugin_tools: - self.tool_registry = {**self.tool_registry, **plugin_tools} + registry = {**registry, **plugin_tools} + virtual_tools = self.plugin_runtime.register_virtual_tools(registry) + if virtual_tools: + registry = {**registry, **virtual_tools} + self.tool_registry = registry self.client = OpenAICompatClient(self.model_config) self.tool_context = build_tool_context(self.runtime_config) @@ -96,6 +105,7 @@ class LocalCodingAgent: self.last_run_result = None self.active_session_id = None self.last_session_path = None + self.resume_source_session_id = None def build_prompt_context(self, scratchpad_directory: Path | None = None): return build_prompt_context( @@ -133,6 +143,7 @@ class LocalCodingAgent: def run(self, prompt: str) -> AgentRunResult: self.managed_agent_id = None + self.resume_source_session_id = None session_id = uuid4().hex scratchpad_directory = self._ensure_scratchpad_directory(session_id) result = self._run_prompt( @@ -147,6 +158,7 @@ class LocalCodingAgent: def resume(self, prompt: str, stored_session: StoredAgentSession) -> AgentRunResult: self.managed_agent_id = None + self.resume_source_session_id = stored_session.session_id session = AgentSessionState.from_persisted( system_prompt_parts=stored_session.system_prompt_parts, user_context=stored_session.user_context, @@ -207,7 +219,8 @@ class LocalCodingAgent: parent_agent_id=self.parent_agent_id, group_id=self.managed_group_id, child_index=self.managed_child_index, - label='root' if base_session is None else 'resume', + label=self.managed_label or ('root' if base_session is None else 'resume'), + resumed_from_session_id=self.resume_source_session_id, ) session = ( base_session @@ -521,6 +534,17 @@ class LocalCodingAgent: 'message_id': session.messages[tool_message_index].message_id, } ) + plugin_preflight_messages = self._plugin_tool_preflight_messages(tool_call.name) + if plugin_preflight_messages: + stream_events.append( + { + 'type': 'plugin_tool_preflight', + 'tool_name': tool_call.name, + 'tool_call_id': tool_call.id, + 'message_id': session.messages[tool_message_index].message_id, + 'message_count': len(plugin_preflight_messages), + } + ) plugin_block_message = self._plugin_block_message(tool_call.name) if plugin_block_message is not None: tool_result = ToolExecutionResult( @@ -592,38 +616,12 @@ class LocalCodingAgent: 'message': message, } ) - plugin_runtime_message = self._build_plugin_tool_runtime_message( - tool_name=tool_call.name, - block_message=plugin_block_message, - plugin_messages=plugin_messages, - ) - if plugin_runtime_message is not None: - session.append_user( - plugin_runtime_message, - metadata={ - 'kind': 'plugin_tool_runtime', - 'tool_name': tool_call.name, - 'tool_call_id': tool_call.id, - 'plugin_blocked': plugin_block_message is not None, - 'plugin_message_count': len(plugin_messages), - }, - message_id=f'plugin_tool_runtime_{tool_call.id}', - ) - stream_events.append( - { - 'type': 'plugin_tool_context', - 'tool_name': tool_call.name, - 'tool_call_id': tool_call.id, - 'message_id': f'plugin_tool_runtime_{tool_call.id}', - 'blocked': plugin_block_message is not None, - 'message_count': len(plugin_messages), - } - ) session.finalize_tool( tool_message_index, content=serialize_tool_result(tool_result), metadata={ 'phase': 'completed', + 'plugin_preflight_messages': list(plugin_preflight_messages), **dict(tool_result.metadata), }, stop_reason='tool_completed', @@ -638,6 +636,41 @@ class LocalCodingAgent: 'metadata': dict(tool_result.metadata), } ) + self._append_runtime_tool_followup_events( + stream_events, + tool_call=tool_call, + tool_result=tool_result, + ) + plugin_runtime_message = self._build_plugin_tool_runtime_message( + tool_name=tool_call.name, + preflight_messages=plugin_preflight_messages, + block_message=plugin_block_message, + plugin_messages=plugin_messages, + ) + if plugin_runtime_message is not None: + session.append_user( + plugin_runtime_message, + metadata={ + 'kind': 'plugin_tool_runtime', + 'tool_name': tool_call.name, + 'tool_call_id': tool_call.id, + 'plugin_blocked': plugin_block_message is not None, + 'plugin_message_count': len(plugin_messages), + 'plugin_preflight_count': len(plugin_preflight_messages), + }, + message_id=f'plugin_tool_runtime_{tool_call.id}', + ) + stream_events.append( + { + 'type': 'plugin_tool_context', + 'tool_name': tool_call.name, + 'tool_call_id': tool_call.id, + 'message_id': f'plugin_tool_runtime_{tool_call.id}', + 'blocked': plugin_block_message is not None, + 'message_count': len(plugin_messages), + 'preflight_count': len(plugin_preflight_messages), + } + ) history_entry = self._build_file_history_entry( tool_call=tool_call, tool_result=tool_result, @@ -980,6 +1013,11 @@ class LocalCodingAgent: 'original_token_estimate': original_tokens, 'replacement_token_estimate': replacement_tokens, 'snipped_turn_index': turn_index, + 'snipped_from_role': message.role, + 'snipped_from_message_id': message.message_id, + 'snipped_from_kind': message.metadata.get('kind'), + 'snipped_from_lineage_id': message.metadata.get('lineage_id'), + 'snipped_from_revision': message.metadata.get('revision'), }, ) delta = original_tokens - replacement_tokens @@ -1024,6 +1062,7 @@ class LocalCodingAgent: if compact_end <= prefix_count: return False candidates = session.messages[prefix_count:compact_end] + preserved_tail = list(session.messages[compact_end:]) if not candidates: return False compacted_tokens = sum( @@ -1048,6 +1087,7 @@ class LocalCodingAgent: estimated_tokens_before=usage_total, estimated_tokens_removed=compacted_tokens, preserved_tail_count=tail_count, + preserved_tail=preserved_tail, ) session.messages = ( session.messages[:prefix_count] @@ -1062,6 +1102,11 @@ class LocalCodingAgent: 'estimated_tokens_before': usage_total, 'estimated_tokens_removed': compacted_tokens, 'preserved_tail_count': tail_count, + 'preserved_tail_ids': [ + message.message_id for message in preserved_tail if message.message_id + ], + 'compaction_depth': compact_message.metadata.get('compaction_depth'), + 'nested_compaction_count': compact_message.metadata.get('nested_compaction_count'), 'compacted_message_ids': [ message.message_id for message in candidates if message.message_id ], @@ -1183,6 +1228,7 @@ class LocalCodingAgent: estimated_tokens_before: int, estimated_tokens_removed: int, preserved_tail_count: int, + preserved_tail, ): summary_lines = [ '', @@ -1215,17 +1261,65 @@ class LocalCodingAgent: ) from .agent_session import AgentMessage + nested_compaction_count = sum( + 1 for message in messages if message.metadata.get('kind') == 'compact_boundary' + ) + prior_depths = [ + int(message.metadata.get('compaction_depth', 0)) + for message in messages + if isinstance(message.metadata.get('compaction_depth', 0), int) + ] + compaction_depth = (max(prior_depths) if prior_depths else 0) + 1 + compacted_kinds: dict[str, int] = {} + compacted_lineage_ids: list[str] = [] + preserved_tail_lineage_ids = [ + lineage_id + for lineage_id in ( + message.metadata.get('lineage_id') for message in preserved_tail + ) + if isinstance(lineage_id, str) and lineage_id + ] + max_source_revision = 0 + compacted_revision_total = 0 + for message in messages: + kind = message.metadata.get('kind') + label = str(kind) if isinstance(kind, str) and kind else message.role + compacted_kinds[label] = compacted_kinds.get(label, 0) + 1 + lineage_id = message.metadata.get('lineage_id') + if isinstance(lineage_id, str) and lineage_id: + compacted_lineage_ids.append(lineage_id) + revision = message.metadata.get('revision') + if isinstance(revision, int) and not isinstance(revision, bool): + max_source_revision = max(max_source_revision, revision) + compacted_revision_total += revision + + compact_boundary_id = f'compact_boundary_{turn_index}_{len(messages)}' + return AgentMessage( role='system', content='\n'.join(summary_lines), - message_id=f'compact_boundary_{turn_index}_{len(messages)}', + message_id=compact_boundary_id, metadata={ 'kind': 'compact_boundary', + 'lineage_id': compact_boundary_id, + 'revision': 0, + 'revision_count': 1, + 'message_role': 'system', 'turn_index': turn_index, 'compacted_message_count': len(messages), 'estimated_tokens_before': estimated_tokens_before, 'estimated_tokens_removed': estimated_tokens_removed, 'preserved_tail_count': preserved_tail_count, + 'preserved_tail_ids': [ + message.message_id for message in preserved_tail if message.message_id + ], + 'compaction_depth': compaction_depth, + 'nested_compaction_count': nested_compaction_count, + 'compacted_kinds': compacted_kinds, + 'compacted_lineage_ids': compacted_lineage_ids, + 'preserved_tail_lineage_ids': preserved_tail_lineage_ids, + 'max_source_revision': max_source_revision, + 'compacted_revision_total': compacted_revision_total, 'compacted_message_ids': [ message.message_id for message in messages if message.message_id ], @@ -1298,6 +1392,7 @@ class LocalCodingAgent: failed_children = 0 child_result = None for index, subtask in enumerate(subtasks, start=1): + subtask_label = str(subtask.get('label') or f'subtask_{index}') child_agent = LocalCodingAgent( model_config=self.model_config, runtime_config=replace( @@ -1312,6 +1407,7 @@ class LocalCodingAgent: parent_agent_id=self.managed_agent_id, managed_group_id=group_id, managed_child_index=index, + managed_label=subtask_label, ) if group_id is not None and child_agent.managed_agent_id is not None: self.agent_manager.register_group_child( @@ -1322,7 +1418,49 @@ class LocalCodingAgent: child_prompt = str(subtask['prompt']) if include_parent_context and prior_results: child_prompt = self._prepend_delegate_context(child_prompt, prior_results) - child_result = child_agent.run(child_prompt) + resume_session_id = subtask.get('resume_session_id') + resume_used = False + if isinstance(resume_session_id, str) and resume_session_id: + try: + stored_child_session = load_agent_session( + resume_session_id, + directory=child_runtime_config.session_directory, + ) + except OSError: + child_result = AgentRunResult( + final_output=f'Unable to load delegated session {resume_session_id}.', + turns=0, + tool_calls=0, + transcript=(), + stop_reason='resume_load_error', + session_id=resume_session_id, + ) + failed_children += 1 + summary = { + 'index': index, + 'label': subtask_label, + 'session_id': resume_session_id, + 'turns': child_result.turns, + 'tool_calls': child_result.tool_calls, + 'stop_reason': child_result.stop_reason or 'resume_load_error', + 'output_preview': self._preview_text(child_result.final_output, 220), + 'resume_used': True, + 'resumed_from_session_id': resume_session_id, + } + child_summaries.append(summary) + prior_results.append( + { + 'label': summary['label'], + 'output_preview': str(summary['output_preview']), + } + ) + if not continue_on_error: + break + continue + child_result = child_agent.resume(child_prompt, stored_child_session) + resume_used = True + else: + child_result = child_agent.run(child_prompt) if group_id is not None and child_agent.managed_agent_id is not None: self.agent_manager.register_group_child( group_id, @@ -1331,12 +1469,18 @@ class LocalCodingAgent: ) summary = { 'index': index, - 'label': str(subtask.get('label') or f'subtask_{index}'), + 'label': subtask_label, 'session_id': child_result.session_id or '', 'turns': child_result.turns, 'tool_calls': child_result.tool_calls, 'stop_reason': child_result.stop_reason or 'stop', 'output_preview': self._preview_text(child_result.final_output, 220), + 'resume_used': resume_used, + 'resumed_from_session_id': ( + str(resume_session_id) + if isinstance(resume_session_id, str) and resume_session_id + else '' + ), } child_summaries.append(summary) if child_result.session_id: @@ -1353,6 +1497,9 @@ class LocalCodingAgent: break assert child_result is not None completed_children = len(child_summaries) - failed_children + resumed_children = sum( + 1 for summary in child_summaries if summary.get('resume_used') + ) group_status = 'completed' if failed_children and completed_children: group_status = 'partial' @@ -1375,6 +1522,7 @@ class LocalCodingAgent: if group_id is not None: summary_lines.append(f'group_id={group_id}') summary_lines.append(f'group_status={group_status}') + summary_lines.append(f'resumed_children={resumed_children}') summary_lines.append('') for summary in child_summaries: summary_lines.extend( @@ -1384,6 +1532,8 @@ class LocalCodingAgent: f"turns={summary['turns']}", f"tool_calls={summary['tool_calls']}", f"stop_reason={summary['stop_reason']}", + f"resume_used={summary['resume_used']}", + f"resumed_from_session_id={summary['resumed_from_session_id']}", f"output_preview={summary['output_preview']}", '', ] @@ -1407,6 +1557,7 @@ class LocalCodingAgent: 'group_status': group_status, 'failed_children': failed_children, 'completed_children': completed_children, + 'resumed_children': resumed_children, }, ) @@ -1431,13 +1582,24 @@ class LocalCodingAgent: 'prompt': prompt.strip(), 'label': label if isinstance(label, str) and label.strip() else f'subtask_{index}', } + resume_session_id = item.get('resume_session_id') + if resume_session_id is None: + resume_session_id = item.get('session_id') + if isinstance(resume_session_id, str) and resume_session_id.strip(): + task['resume_session_id'] = resume_session_id.strip() if isinstance(max_turns, int) and not isinstance(max_turns, bool) and max_turns > 0: task['max_turns'] = max_turns subtasks.append(task) prompt = arguments.get('prompt') if isinstance(prompt, str) and prompt.strip(): if not subtasks: - subtasks.append({'prompt': prompt.strip(), 'label': 'subtask_1'}) + task: dict[str, object] = {'prompt': prompt.strip(), 'label': 'subtask_1'} + resume_session_id = arguments.get('resume_session_id') + if resume_session_id is None: + resume_session_id = arguments.get('session_id') + if isinstance(resume_session_id, str) and resume_session_id.strip(): + task['resume_session_id'] = resume_session_id.strip() + subtasks.append(task) return subtasks[:8] def _delegated_task_units( @@ -1476,6 +1638,60 @@ class LocalCodingAgent: lines.extend(['', '', prompt]) return '\n'.join(lines) + def _append_runtime_tool_followup_events( + self, + stream_events: list[dict[str, object]], + *, + tool_call: ToolCall, + tool_result: ToolExecutionResult, + ) -> None: + metadata = tool_result.metadata + if metadata.get('action') == 'plugin_virtual_tool': + stream_events.append( + { + 'type': 'plugin_virtual_tool_result', + 'tool_call_id': tool_call.id, + 'tool_name': tool_call.name, + 'plugin_name': metadata.get('plugin_name'), + 'virtual_tool': metadata.get('virtual_tool'), + } + ) + if tool_call.name != 'delegate_agent': + return + child_results = metadata.get('child_results') + if isinstance(child_results, list): + for child in child_results: + if not isinstance(child, dict): + continue + stream_events.append( + { + 'type': 'delegate_subtask_result', + 'tool_call_id': tool_call.id, + 'group_id': metadata.get('group_id'), + 'label': child.get('label'), + 'index': child.get('index'), + 'session_id': child.get('session_id'), + 'stop_reason': child.get('stop_reason'), + 'tool_calls': child.get('tool_calls'), + 'turns': child.get('turns'), + 'resume_used': child.get('resume_used'), + 'resumed_from_session_id': child.get('resumed_from_session_id'), + } + ) + if metadata.get('group_id') is not None: + stream_events.append( + { + 'type': 'delegate_group_result', + 'tool_call_id': tool_call.id, + 'group_id': metadata.get('group_id'), + 'group_status': metadata.get('group_status'), + 'subtask_count': metadata.get('subtask_count'), + 'completed_children': metadata.get('completed_children'), + 'failed_children': metadata.get('failed_children'), + 'resumed_children': metadata.get('resumed_children'), + } + ) + def _preview_text(self, text: str, limit: int) -> str: normalized = ' '.join(text.split()) if len(normalized) <= limit: @@ -1597,12 +1813,31 @@ class LocalCodingAgent: lines.append( f"- Latest compact boundary id: {latest_boundary.message_id or '(none)'}" ) + depth = latest_boundary.metadata.get('compaction_depth') + if isinstance(depth, int) and not isinstance(depth, bool): + lines.append(f'- Latest compaction depth: {depth}') + compacted_lineages = latest_boundary.metadata.get('compacted_lineage_ids') + if isinstance(compacted_lineages, list) and compacted_lineages: + lines.append(f'- Latest compacted lineages: {len(compacted_lineages)}') + preserved_tail = latest_boundary.metadata.get('preserved_tail_ids') + if isinstance(preserved_tail, list) and preserved_tail: + lines.append( + '- Latest preserved tail ids: ' + + ', '.join(str(item) for item in preserved_tail[:4]) + ) if snipped_messages: last_ids = [ message.message_id or '(none)' for message in snipped_messages[-3:] ] lines.append(f"- Recent snipped ids: {', '.join(last_ids)}") + snipped_lineages = [ + str(message.metadata.get('snipped_from_lineage_id')) + for message in snipped_messages[-3:] + if isinstance(message.metadata.get('snipped_from_lineage_id'), str) + ] + if snipped_lineages: + lines.append(f"- Recent snipped lineages: {', '.join(snipped_lineages)}") lines.extend( [ '', @@ -1616,19 +1851,22 @@ class LocalCodingAgent: self, *, tool_name: str, + preflight_messages: tuple[str, ...], block_message: str | None, plugin_messages: tuple[str, ...], ) -> str | None: - if block_message is None and not plugin_messages: + if block_message is None and not plugin_messages and not preflight_messages: return None lines = [ '', f'Plugin tool runtime guidance for `{tool_name}`:', ] + for message in preflight_messages: + lines.append(f'- Before tool: {message}') if block_message is not None: lines.append(f'- Blocked: {block_message}') for message in plugin_messages: - lines.append(f'- {message}') + lines.append(f'- After result: {message}') lines.extend( [ '', @@ -1638,6 +1876,11 @@ class LocalCodingAgent: ) return '\n'.join(lines) + def _plugin_tool_preflight_messages(self, tool_name: str) -> tuple[str, ...]: + if self.plugin_runtime is None: + return () + return self.plugin_runtime.tool_preflight_injections(tool_name) + def _plugin_block_message(self, tool_name: str) -> str | None: if self.plugin_runtime is None: return None @@ -1763,6 +2006,7 @@ class LocalCodingAgent: def _finalize_managed_agent(self, result: AgentRunResult) -> None: if self.managed_agent_id is None or self.agent_manager is None: + self.resume_source_session_id = None return self.agent_manager.finish_agent( self.managed_agent_id, @@ -1772,6 +2016,7 @@ class LocalCodingAgent: tool_calls=result.tool_calls, stop_reason=result.stop_reason, ) + self.resume_source_session_id = None def _apply_plugin_before_prompt_hooks(self, prompt: str) -> str: if self.plugin_runtime is None: diff --git a/src/agent_session.py b/src/agent_session.py index 02a204a..3c7e290 100644 --- a/src/agent_session.py +++ b/src/agent_session.py @@ -114,6 +114,10 @@ class AgentSessionState: _append_system_context(system_prompt_parts, state.system_context) ), blocks=_text_blocks('\n\n'.join(_append_system_context(system_prompt_parts, state.system_context))), + metadata=_initialize_message_metadata( + role='system', + message_id='system_0', + ), ) ) if state.user_context: @@ -122,6 +126,10 @@ class AgentSessionState: role='user', content=_render_user_context_reminder(state.user_context), blocks=_text_blocks(_render_user_context_reminder(state.user_context)), + metadata=_initialize_message_metadata( + role='user', + message_id='user_context_0', + ), ) ) if user_prompt is not None: @@ -130,6 +138,10 @@ class AgentSessionState: role='user', content=user_prompt, blocks=_text_blocks(user_prompt), + metadata=_initialize_message_metadata( + role='user', + message_id='user_0', + ), ) ) return state @@ -152,6 +164,10 @@ class AgentSessionState: message_id=message_id, stop_reason=stop_reason, usage=usage or UsageStats(), + metadata=_initialize_message_metadata( + role='assistant', + message_id=message_id or f'assistant_{len(self.messages)}', + ), ) ) @@ -168,16 +184,29 @@ class AgentSessionState: blocks=(), message_id=message_id, state='streaming', + metadata=_initialize_message_metadata( + role='assistant', + message_id=message_id or f'assistant_{len(self.messages)}', + ), ) ) return len(self.messages) - 1 def append_assistant_delta(self, index: int, delta: str) -> None: message = self.messages[index] + merged_metadata = _record_mutation( + dict(message.metadata), + mutation_kind='assistant_delta_append', + previous_content=message.content, + previous_state=message.state, + previous_stop_reason=message.stop_reason, + ) + merged_metadata = _advance_lineage_revision(merged_metadata) self.messages[index] = replace( message, content=message.content + delta, blocks=_assistant_blocks(message.content + delta, message.tool_calls), + metadata=merged_metadata, ) def merge_assistant_tool_call_delta( @@ -211,10 +240,19 @@ class AgentSessionState: if arguments_delta: current_arguments = function_block.get('arguments', '') function_block['arguments'] = f'{current_arguments}{arguments_delta}' + merged_metadata = _record_mutation( + dict(message.metadata), + mutation_kind='assistant_tool_call_delta', + previous_content=message.content, + previous_state=message.state, + previous_stop_reason=message.stop_reason, + ) + merged_metadata = _advance_lineage_revision(merged_metadata) self.messages[index] = replace( message, tool_calls=tuple(tool_calls), blocks=_assistant_blocks(message.content, tuple(tool_calls)), + metadata=merged_metadata, ) def finalize_assistant( @@ -225,12 +263,21 @@ class AgentSessionState: usage: UsageStats | None = None, ) -> None: message = self.messages[index] + merged_metadata = _record_mutation( + dict(message.metadata), + mutation_kind='assistant_finalize', + previous_content=message.content, + previous_state=message.state, + previous_stop_reason=message.stop_reason, + ) + merged_metadata = _advance_lineage_revision(merged_metadata) self.messages[index] = replace( message, state='final', stop_reason=finish_reason, usage=usage or message.usage, blocks=_assistant_blocks(message.content, message.tool_calls), + metadata=merged_metadata, ) def append_user( @@ -245,7 +292,11 @@ class AgentSessionState: role='user', content=content, blocks=_text_blocks(content), - metadata=dict(metadata or {}), + metadata=_initialize_message_metadata( + role='user', + message_id=message_id or f'user_{len(self.messages)}', + metadata=dict(metadata or {}), + ), message_id=message_id, ) ) @@ -258,6 +309,11 @@ class AgentSessionState: name=name, tool_call_id=tool_call_id, blocks=_tool_blocks(name, tool_call_id, content), + metadata=_initialize_message_metadata( + role='tool', + message_id=f'tool_{len(self.messages)}', + metadata={'tool_name': name, 'tool_call_id': tool_call_id}, + ), ) ) @@ -278,7 +334,15 @@ class AgentSessionState: blocks=(), message_id=message_id, state='streaming', - metadata=dict(metadata or {}), + metadata=_initialize_message_metadata( + role='tool', + message_id=message_id or f'tool_{len(self.messages)}', + metadata={ + 'tool_name': name, + 'tool_call_id': tool_call_id, + **dict(metadata or {}), + }, + ), ) ) return len(self.messages) - 1 @@ -292,6 +356,14 @@ class AgentSessionState: ) -> None: message = self.messages[index] merged_metadata = dict(message.metadata) + merged_metadata = _record_mutation( + merged_metadata, + mutation_kind='tool_delta_append', + previous_content=message.content, + previous_state=message.state, + previous_stop_reason=message.stop_reason, + ) + merged_metadata = _advance_lineage_revision(merged_metadata) if metadata: merged_metadata.update(metadata) self.messages[index] = replace( @@ -320,6 +392,7 @@ class AgentSessionState: previous_state=message.state, previous_stop_reason=message.stop_reason, ) + merged_metadata = _advance_lineage_revision(merged_metadata) if metadata: merged_metadata.update(metadata) self.messages[index] = replace( @@ -358,6 +431,7 @@ class AgentSessionState: previous_state=message.state, previous_stop_reason=message.stop_reason, ) + merged_metadata = _advance_lineage_revision(merged_metadata) if metadata: merged_metadata.update(metadata) self.messages[index] = replace( @@ -471,9 +545,64 @@ def _record_mutation( metadata['mutations'] = mutations metadata['mutation_count'] = len(mutations) metadata['last_mutation_kind'] = mutation_kind + totals = metadata.get('mutation_totals') + if not isinstance(totals, dict): + totals = {} + else: + totals = { + str(key): int(value) + for key, value in totals.items() + if isinstance(key, str) and not isinstance(value, bool) and isinstance(value, int) + } + totals[mutation_kind] = totals.get(mutation_kind, 0) + 1 + metadata['mutation_totals'] = totals return metadata +def _initialize_message_metadata( + *, + role: str, + message_id: str | None, + metadata: JSONDict | None = None, +) -> JSONDict: + merged = dict(metadata or {}) + lineage_id = merged.get('lineage_id') + if not isinstance(lineage_id, str) or not lineage_id: + if isinstance(message_id, str) and message_id: + lineage_id = message_id + else: + lineage_id = f'{role}_lineage' + revision = merged.get('revision') + if isinstance(revision, bool) or not isinstance(revision, int): + revision = 0 + revision_count = merged.get('revision_count') + if isinstance(revision_count, bool) or not isinstance(revision_count, int): + revision_count = max(revision + 1, 1) + merged['lineage_id'] = lineage_id + merged['revision'] = revision + merged['revision_count'] = revision_count + merged.setdefault('message_role', role) + return merged + + +def _advance_lineage_revision(metadata: JSONDict) -> JSONDict: + normalized = _initialize_message_metadata( + role=str(metadata.get('message_role', 'message')), + message_id=metadata.get('lineage_id') if isinstance(metadata.get('lineage_id'), str) else None, + metadata=metadata, + ) + revision = normalized.get('revision', 0) + if isinstance(revision, bool) or not isinstance(revision, int): + revision = 0 + revision += 1 + normalized['revision'] = revision + revision_count = normalized.get('revision_count', 1) + if isinstance(revision_count, bool) or not isinstance(revision_count, int): + revision_count = 1 + normalized['revision_count'] = max(revision_count, revision + 1) + return normalized + + def _text_blocks(text: str) -> tuple[JSONDict, ...]: if not text: return () diff --git a/src/agent_tools.py b/src/agent_tools.py index 6e810d6..96990fd 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -240,16 +240,21 @@ def default_tool_registry() -> dict[str, AgentTool]: 'prompt': {'type': 'string'}, 'label': {'type': 'string'}, 'max_turns': {'type': 'integer', 'minimum': 1, 'maximum': 20}, + 'resume_session_id': {'type': 'string'}, + 'session_id': {'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'}, }, }, handler=_delegate_agent_placeholder, diff --git a/src/plugin_runtime.py b/src/plugin_runtime.py index 512929a..a61f71e 100644 --- a/src/plugin_runtime.py +++ b/src/plugin_runtime.py @@ -16,10 +16,20 @@ class PluginToolAlias: @dataclass(frozen=True) class PluginToolHook: tool_name: str + before_tool: str | None = None after_result: str | None = None block_message: str | None = None +@dataclass(frozen=True) +class PluginVirtualTool: + name: str + description: str + response_template: str + parameters: dict[str, Any] = field(default_factory=dict) + metadata: dict[str, Any] = field(default_factory=dict) + + @dataclass(frozen=True) class PluginManifest: name: str @@ -29,6 +39,7 @@ class PluginManifest: tool_names: tuple[str, ...] = () hook_names: tuple[str, ...] = () tool_aliases: tuple[PluginToolAlias, ...] = () + virtual_tools: tuple[PluginVirtualTool, ...] = () tool_hooks: tuple[PluginToolHook, ...] = () blocked_tools: tuple[str, ...] = () before_prompt: str | None = None @@ -69,6 +80,11 @@ class PluginRuntime: 'Tool aliases: ' + ', '.join(alias.name for alias in manifest.tool_aliases) ) + if manifest.virtual_tools: + lines.append( + 'Virtual tools: ' + + ', '.join(tool.name for tool in manifest.virtual_tools) + ) if manifest.blocked_tools: lines.append( 'Blocked tools: ' @@ -114,6 +130,26 @@ class PluginRuntime: ) return aliases + def register_virtual_tools( + self, + base_registry: dict[str, Any] | None = None, + ) -> dict[str, Any]: + from .agent_tools import AgentTool + + tools: dict[str, AgentTool] = {} + occupied = set(base_registry or {}) + for manifest in self.manifests: + for tool in manifest.virtual_tools: + if tool.name in occupied or tool.name in tools: + continue + tools[tool.name] = AgentTool( + name=tool.name, + description=tool.description, + parameters=tool.parameters or {'type': 'object', 'properties': {}}, + handler=_build_virtual_tool_handler(manifest.name, tool), + ) + return tools + def blocked_tool_message(self, tool_name: str) -> str | None: for manifest in self.manifests: if tool_name in manifest.blocked_tools: @@ -123,6 +159,14 @@ class PluginRuntime: return hook.block_message return None + def tool_preflight_injections(self, tool_name: str) -> tuple[str, ...]: + messages: list[str] = [] + for manifest in self.manifests: + for hook in manifest.tool_hooks: + if hook.tool_name == tool_name and hook.before_tool: + messages.append(f'{manifest.name}: {hook.before_tool}') + return tuple(messages) + def tool_result_injections(self, tool_name: str) -> tuple[str, ...]: messages: list[str] = [] for manifest in self.manifests: @@ -145,6 +189,8 @@ class PluginRuntime: details.append(f'hooks={len(manifest.hook_names)}') if manifest.tool_aliases: details.append(f'aliases={len(manifest.tool_aliases)}') + if manifest.virtual_tools: + details.append(f'virtual_tools={len(manifest.virtual_tools)}') if manifest.blocked_tools: details.append(f'blocked={len(manifest.blocked_tools)}') if manifest.tool_hooks: @@ -211,6 +257,7 @@ def _load_manifest(path: Path) -> PluginManifest | None: tool_names=_extract_string_tuple(payload.get('tools')), hook_names=hook_names, tool_aliases=_extract_tool_aliases(payload), + virtual_tools=_extract_virtual_tools(payload), tool_hooks=_extract_tool_hooks(payload), blocked_tools=_extract_string_tuple( payload.get('blocked_tools') @@ -303,6 +350,9 @@ def _extract_tool_hooks(payload: dict[str, Any]) -> tuple[PluginToolHook, ...]: continue if not isinstance(value, dict): continue + before_tool = value.get('beforeTool') + if before_tool is None: + before_tool = value.get('before_tool') after_result = value.get('afterResult') if after_result is None: after_result = value.get('after_result') @@ -312,8 +362,80 @@ def _extract_tool_hooks(payload: dict[str, Any]) -> tuple[PluginToolHook, ...]: hooks.append( PluginToolHook( tool_name=tool_name.strip(), + before_tool=_optional_string(before_tool), after_result=_optional_string(after_result), block_message=_optional_string(block_message), ) ) return tuple(hooks) + + +def _extract_virtual_tools(payload: dict[str, Any]) -> tuple[PluginVirtualTool, ...]: + raw_tools = payload.get('virtual_tools') + if raw_tools is None: + raw_tools = payload.get('virtualTools') + if raw_tools is None: + raw_tools = payload.get('runtimeTools') + tools: list[PluginVirtualTool] = [] + if isinstance(raw_tools, list): + for item in raw_tools: + if not isinstance(item, dict): + continue + name = _optional_string(item.get('name')) + description = _optional_string(item.get('description')) + response_template = item.get('responseTemplate') + if response_template is None: + response_template = item.get('response_template') + if response_template is None: + response_template = item.get('response') + response_text = _optional_string(response_template) + parameters = item.get('parameters') + metadata = item.get('metadata') + if not name or not description or not response_text: + continue + tools.append( + PluginVirtualTool( + name=name, + description=description, + response_template=response_text, + parameters=dict(parameters) if isinstance(parameters, dict) else {}, + metadata=dict(metadata) if isinstance(metadata, dict) else {}, + ) + ) + return tuple(tools) + + +def _build_virtual_tool_handler( + plugin_name: str, + tool: PluginVirtualTool, +): + def _handler(arguments: dict[str, Any], context): # noqa: ANN001 + rendered = _render_virtual_tool_response(tool.response_template, arguments) + metadata = { + 'action': 'plugin_virtual_tool', + 'plugin_name': plugin_name, + 'virtual_tool': tool.name, + **tool.metadata, + } + return rendered, metadata + + return _handler + + +def _render_virtual_tool_response( + template: str, + arguments: dict[str, Any], +) -> str: + normalized = { + key: json.dumps(value, ensure_ascii=True) if isinstance(value, (dict, list)) else str(value) + for key, value in arguments.items() + } + try: + return template.format_map(_SafeTemplateDict(normalized)) + except Exception: + return template + + +class _SafeTemplateDict(dict[str, str]): + def __missing__(self, key: str) -> str: + return '{' + key + '}' diff --git a/src/query_engine.py b/src/query_engine.py index d018a07..06e1013 100644 --- a/src/query_engine.py +++ b/src/query_engine.py @@ -55,7 +55,17 @@ class QueryEnginePort: plugin_runtime: PluginRuntime | None = None runtime_event_counts: dict[str, int] = field(default_factory=dict) runtime_message_kind_counts: dict[str, int] = field(default_factory=dict) + runtime_mutation_counts: dict[str, int] = field(default_factory=dict) + runtime_group_status_counts: dict[str, int] = field(default_factory=dict) + runtime_child_stop_reason_counts: dict[str, int] = field(default_factory=dict) + runtime_resumed_children: int = 0 + runtime_context_reduction: dict[str, int] = field(default_factory=dict) + runtime_lineage_stats: dict[str, int] = field(default_factory=dict) runtime_transcript_size: int = 0 + _runtime_seen_lineages: set[str] = field(default_factory=set, init=False, repr=False) + _runtime_revised_lineages: set[str] = field(default_factory=set, init=False, repr=False) + _runtime_tombstoned_lineages: set[str] = field(default_factory=set, init=False, repr=False) + _runtime_compacted_lineages: set[str] = field(default_factory=set, init=False, repr=False) last_turn: TurnResult | None = field(default=None, init=False, repr=False) @classmethod @@ -180,6 +190,7 @@ class QueryEnginePort: if self.config.use_runtime_agent: for event in result.events: yield event + yield self._runtime_summary_event() yield { 'type': 'message_stop', 'usage': { @@ -217,7 +228,7 @@ class QueryEnginePort: path = save_session( StoredSession( session_id=self.session_id, - messages=tuple(self.mutable_messages), + messages=self.transcript_store.replay(), input_tokens=self.total_usage.input_tokens, output_tokens=self.total_usage.output_tokens, ) @@ -247,6 +258,7 @@ class QueryEnginePort: f'Transcript flushed: {self.transcript_store.flushed}', f'Real runtime agent mode: {self.config.use_runtime_agent}', ] + sections.extend(['', '## Transcript Store', *self.transcript_store.summary_lines()]) if self.plugin_runtime is not None: sections.extend(['', '## Plugin Runtime', self.plugin_runtime.render_summary()]) if self.runtime_agent is not None and self.runtime_agent.agent_manager is not None: @@ -264,6 +276,38 @@ class QueryEnginePort: f'- {name}={count}' for name, count in sorted(self.runtime_message_kind_counts.items()) ) + if self.runtime_mutation_counts: + sections.extend(['', '## Runtime Mutations']) + sections.extend( + f'- {name}={count}' + for name, count in sorted(self.runtime_mutation_counts.items()) + ) + if self.runtime_group_status_counts or self.runtime_child_stop_reason_counts: + sections.extend(['', '## Runtime Orchestration']) + if self.runtime_group_status_counts: + sections.extend( + f'- group_status:{name}={count}' + for name, count in sorted(self.runtime_group_status_counts.items()) + ) + if self.runtime_child_stop_reason_counts: + sections.extend( + f'- child_stop:{name}={count}' + for name, count in sorted(self.runtime_child_stop_reason_counts.items()) + ) + if self.runtime_resumed_children: + sections.append(f'- resumed_children={self.runtime_resumed_children}') + if self.runtime_context_reduction: + sections.extend(['', '## Runtime Context Reduction']) + sections.extend( + f'- {name}={count}' + for name, count in sorted(self.runtime_context_reduction.items()) + ) + if self.runtime_lineage_stats: + sections.extend(['', '## Runtime Lineage']) + sections.extend( + f'- {name}={count}' + for name, count in sorted(self.runtime_lineage_stats.items()) + ) if self.last_turn is not None: sections.extend( [ @@ -303,8 +347,8 @@ class QueryEnginePort: denied_tools: tuple[PermissionDenial, ...], ) -> None: self.mutable_messages.append(prompt) - self.transcript_store.append(prompt) - self.transcript_store.append(turn.output) + self.transcript_store.append(prompt, kind='prompt') + self.transcript_store.append(turn.output, kind='output') if self.config.use_runtime_agent: self._record_runtime_turn(turn) self.permission_denials.extend(denied_tools) @@ -334,7 +378,22 @@ class QueryEnginePort: self.runtime_event_counts[event_type] = ( self.runtime_event_counts.get(event_type, 0) + 1 ) + if event_type == 'delegate_group_result': + group_status = event.get('group_status') + if isinstance(group_status, str) and group_status: + self.runtime_group_status_counts[group_status] = ( + self.runtime_group_status_counts.get(group_status, 0) + 1 + ) + elif event_type == 'delegate_subtask_result': + stop_reason = event.get('stop_reason') + if isinstance(stop_reason, str) and stop_reason: + self.runtime_child_stop_reason_counts[stop_reason] = ( + self.runtime_child_stop_reason_counts.get(stop_reason, 0) + 1 + ) + if bool(event.get('resume_used')): + self.runtime_resumed_children += 1 kind_counts: dict[str, int] = {} + mutation_counts: dict[str, int] = {} for entry in turn.transcript: if not isinstance(entry, dict): continue @@ -342,20 +401,76 @@ class QueryEnginePort: if not isinstance(metadata, dict): continue kind = metadata.get('kind') - if not isinstance(kind, str) or not kind: - continue - kind_counts[kind] = kind_counts.get(kind, 0) + 1 - self.runtime_message_kind_counts[kind] = ( - self.runtime_message_kind_counts.get(kind, 0) + 1 - ) - summary = self._summarize_runtime_turn(event_counts, kind_counts, len(turn.transcript)) + if isinstance(kind, str) and kind: + kind_counts[kind] = kind_counts.get(kind, 0) + 1 + self.runtime_message_kind_counts[kind] = ( + self.runtime_message_kind_counts.get(kind, 0) + 1 + ) + mutation_totals = metadata.get('mutation_totals') + if isinstance(mutation_totals, dict): + for mutation_kind, count in mutation_totals.items(): + if ( + not isinstance(mutation_kind, str) + or not mutation_kind + or isinstance(count, bool) + or not isinstance(count, int) + or count <= 0 + ): + continue + mutation_counts[mutation_kind] = mutation_counts.get(mutation_kind, 0) + count + self.runtime_mutation_counts[mutation_kind] = ( + self.runtime_mutation_counts.get(mutation_kind, 0) + count + ) + else: + mutations = metadata.get('mutations') + if isinstance(mutations, list): + for mutation in mutations: + if not isinstance(mutation, dict): + continue + mutation_kind = mutation.get('kind') + if not isinstance(mutation_kind, str) or not mutation_kind: + continue + mutation_counts[mutation_kind] = mutation_counts.get(mutation_kind, 0) + 1 + self.runtime_mutation_counts[mutation_kind] = ( + self.runtime_mutation_counts.get(mutation_kind, 0) + 1 + ) + self._record_context_reduction(metadata) + self._record_lineage(metadata) + summary = self._summarize_runtime_turn( + event_counts, + kind_counts, + mutation_counts, + len(turn.transcript), + ) if summary: - self.transcript_store.append(summary) + self.transcript_store.append( + summary, + kind='runtime_summary', + metadata={ + 'runtime_event_counts': dict(event_counts), + 'runtime_kind_counts': dict(kind_counts), + 'runtime_mutation_counts': dict(mutation_counts), + 'runtime_transcript_size': len(turn.transcript), + }, + ) + if event_counts: + self.transcript_store.append( + json.dumps(event_counts, sort_keys=True), + kind='runtime_events', + metadata={'counts': dict(event_counts)}, + ) + if kind_counts: + self.transcript_store.append( + json.dumps(kind_counts, sort_keys=True), + kind='runtime_message_kinds', + metadata={'counts': dict(kind_counts)}, + ) def _summarize_runtime_turn( self, event_counts: dict[str, int], kind_counts: dict[str, int], + mutation_counts: dict[str, int], transcript_size: int, ) -> str: parts = [f'runtime_transcript={transcript_size}'] @@ -375,4 +490,116 @@ class QueryEnginePort: for name, count in sorted(kind_counts.items()) ) ) + if mutation_counts: + parts.append( + 'mutations=' + + ', '.join( + f'{name}:{count}' + for name, count in sorted(mutation_counts.items()) + ) + ) return '[runtime] ' + ' | '.join(parts) + + def _runtime_summary_event(self) -> dict[str, object]: + return { + 'type': 'runtime_summary', + 'runtime_event_counts': dict(self.runtime_event_counts), + 'runtime_message_kind_counts': dict(self.runtime_message_kind_counts), + 'runtime_mutation_counts': dict(self.runtime_mutation_counts), + 'runtime_group_status_counts': dict(self.runtime_group_status_counts), + 'runtime_child_stop_reason_counts': dict(self.runtime_child_stop_reason_counts), + 'runtime_resumed_children': self.runtime_resumed_children, + 'runtime_context_reduction': dict(self.runtime_context_reduction), + 'runtime_lineage_stats': dict(self.runtime_lineage_stats), + 'runtime_transcript_size': self.runtime_transcript_size, + 'transcript_store_entries': len(self.transcript_store.entries), + 'transcript_store_compactions': self.transcript_store.compaction_count, + } + + def _record_context_reduction(self, metadata: dict[str, object]) -> None: + kind = metadata.get('kind') + if kind == 'compact_boundary': + self.runtime_context_reduction['compact_boundaries'] = ( + self.runtime_context_reduction.get('compact_boundaries', 0) + 1 + ) + depth = metadata.get('compaction_depth') + if isinstance(depth, int) and not isinstance(depth, bool): + current = self.runtime_context_reduction.get('max_compaction_depth', 0) + self.runtime_context_reduction['max_compaction_depth'] = max(current, depth) + nested = metadata.get('nested_compaction_count') + if isinstance(nested, int) and not isinstance(nested, bool): + self.runtime_context_reduction['nested_compaction_count'] = ( + self.runtime_context_reduction.get('nested_compaction_count', 0) + nested + ) + preserved_tail_count = metadata.get('preserved_tail_count') + if isinstance(preserved_tail_count, int) and not isinstance(preserved_tail_count, bool): + self.runtime_context_reduction['preserved_tail_messages'] = ( + self.runtime_context_reduction.get('preserved_tail_messages', 0) + + preserved_tail_count + ) + compacted_lineage_ids = metadata.get('compacted_lineage_ids') + if isinstance(compacted_lineage_ids, list): + self.runtime_context_reduction['compacted_lineages'] = ( + self.runtime_context_reduction.get('compacted_lineages', 0) + + len( + [ + lineage_id for lineage_id in compacted_lineage_ids + if isinstance(lineage_id, str) and lineage_id + ] + ) + ) + elif kind == 'snipped_message': + self.runtime_context_reduction['snipped_messages'] = ( + self.runtime_context_reduction.get('snipped_messages', 0) + 1 + ) + revision = metadata.get('snipped_from_revision') + if isinstance(revision, int) and not isinstance(revision, bool) and revision > 0: + self.runtime_context_reduction['snipped_revised_messages'] = ( + self.runtime_context_reduction.get('snipped_revised_messages', 0) + 1 + ) + + def _record_lineage(self, metadata: dict[str, object]) -> None: + lineage_id = metadata.get('lineage_id') + if isinstance(lineage_id, str) and lineage_id: + self._runtime_seen_lineages.add(lineage_id) + revision = metadata.get('revision') + if isinstance(revision, int) and not isinstance(revision, bool): + current = self.runtime_lineage_stats.get('max_revision', 0) + self.runtime_lineage_stats['max_revision'] = max(current, revision) + if revision > 0: + self._runtime_revised_lineages.add(lineage_id) + revision_count = metadata.get('revision_count') + if isinstance(revision_count, int) and not isinstance(revision_count, bool): + current = self.runtime_lineage_stats.get('max_revision_count', 0) + self.runtime_lineage_stats['max_revision_count'] = max(current, revision_count) + + kind = metadata.get('kind') + if kind == 'snipped_message': + source_lineage = metadata.get('snipped_from_lineage_id') + if isinstance(source_lineage, str) and source_lineage: + self._runtime_tombstoned_lineages.add(source_lineage) + elif kind == 'compact_boundary': + compacted_lineage_ids = metadata.get('compacted_lineage_ids') + if isinstance(compacted_lineage_ids, list): + for source_lineage in compacted_lineage_ids: + if isinstance(source_lineage, str) and source_lineage: + self._runtime_compacted_lineages.add(source_lineage) + max_source_revision = metadata.get('max_source_revision') + if ( + isinstance(max_source_revision, int) + and not isinstance(max_source_revision, bool) + ): + current = self.runtime_lineage_stats.get('max_source_revision', 0) + self.runtime_lineage_stats['max_source_revision'] = max( + current, + max_source_revision, + ) + + self.runtime_lineage_stats['seen_lineages'] = len(self._runtime_seen_lineages) + self.runtime_lineage_stats['revised_lineages'] = len(self._runtime_revised_lineages) + self.runtime_lineage_stats['tombstoned_lineages'] = len( + self._runtime_tombstoned_lineages + ) + self.runtime_lineage_stats['compacted_lineages'] = len( + self._runtime_compacted_lineages + ) diff --git a/src/transcript.py b/src/transcript.py index 7a01a3d..936c10f 100644 --- a/src/transcript.py +++ b/src/transcript.py @@ -1,23 +1,131 @@ -from __future__ import annotations - -from dataclasses import dataclass, field - - -@dataclass -class TranscriptStore: - entries: list[str] = field(default_factory=list) - flushed: bool = False - - def append(self, entry: str) -> None: - self.entries.append(entry) - self.flushed = False - - def compact(self, keep_last: int = 10) -> None: - if len(self.entries) > keep_last: - self.entries[:] = self.entries[-keep_last:] - - def replay(self) -> tuple[str, ...]: - return tuple(self.entries) - - def flush(self) -> None: - self.flushed = True +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +JSONDict = dict[str, Any] + + +@dataclass(frozen=True) +class TranscriptEntry: + content: str + kind: str = 'message' + metadata: JSONDict = field(default_factory=dict) + + +@dataclass +class TranscriptStore: + entries: list[TranscriptEntry] = field(default_factory=list) + flushed: bool = False + compaction_count: int = 0 + + def __post_init__(self) -> None: + normalized: list[TranscriptEntry] = [] + for entry in self.entries: + if isinstance(entry, TranscriptEntry): + normalized.append(entry) + elif isinstance(entry, str): + normalized.append(TranscriptEntry(content=entry)) + self.entries = normalized + + def append( + self, + entry: str, + *, + kind: str = 'message', + metadata: JSONDict | None = None, + ) -> None: + self.entries.append( + TranscriptEntry( + content=entry, + kind=kind, + metadata=dict(metadata or {}), + ) + ) + self.flushed = False + + def compact(self, keep_last: int = 10) -> None: + if len(self.entries) <= keep_last: + return + trimmed = self.entries[:-keep_last] + kept = self.entries[-keep_last:] + self.compaction_count += 1 + summary_metadata = { + 'kind': 'compact_summary', + 'compaction_index': self.compaction_count, + 'compacted_entries': len(trimmed), + 'compacted_kinds': self._kind_counts(trimmed), + } + summary = TranscriptEntry( + content=self._render_compaction_summary(trimmed, self.compaction_count), + kind='compact_summary', + metadata=summary_metadata, + ) + self.entries[:] = [summary, *kept] + + def replay(self) -> tuple[str, ...]: + return tuple(entry.content for entry in self.entries) + + def flush(self) -> None: + self.flushed = True + + def summary_lines(self) -> list[str]: + lines = [ + f'- Transcript entries: {len(self.entries)}', + f'- Transcript compactions: {self.compaction_count}', + ] + kind_counts = self._kind_counts(self.entries) + if kind_counts: + lines.append( + '- Transcript kinds: ' + + ', '.join( + f'{name}={count}' for name, count in sorted(kind_counts.items()) + ) + ) + return lines + + def structured_replay(self) -> tuple[JSONDict, ...]: + return tuple( + { + 'kind': entry.kind, + 'content': entry.content, + 'metadata': dict(entry.metadata), + } + for entry in self.entries + ) + + def _render_compaction_summary( + self, + trimmed: list[TranscriptEntry], + compaction_index: int, + ) -> str: + lines = [ + f'[transcript-compaction {compaction_index}]', + f'Compacted {len(trimmed)} older transcript entries.', + ] + kind_counts = self._kind_counts(trimmed) + if kind_counts: + lines.append( + 'Kinds: ' + + ', '.join( + f'{name}={count}' for name, count in sorted(kind_counts.items()) + ) + ) + previews: list[str] = [] + for entry in trimmed[-3:]: + preview = ' '.join(entry.content.split()) + if len(preview) > 72: + preview = preview[:69] + '...' + previews.append(f'{entry.kind}: {preview or "(empty)"}') + if previews: + lines.append('Recent compacted previews:') + lines.extend(f'- {preview}' for preview in previews) + return '\n'.join(lines) + + @staticmethod + def _kind_counts(entries: list[TranscriptEntry]) -> JSONDict: + counts: dict[str, int] = {} + for entry in entries: + counts[entry.kind] = counts.get(entry.kind, 0) + 1 + return counts diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py index 65c718b..66e79f4 100644 --- a/tests/test_agent_runtime.py +++ b/tests/test_agent_runtime.py @@ -362,6 +362,14 @@ class AgentRuntimeTests(unittest.TestCase): self.assertIsNotNone(result.scratchpad_directory) assert result.scratchpad_directory is not None self.assertTrue(Path(result.scratchpad_directory).is_dir()) + assistant_messages = [ + message for message in result.transcript if message.get('role') == 'assistant' + ] + self.assertEqual(len(assistant_messages), 1) + metadata = assistant_messages[0].get('metadata', {}) + mutation_totals = metadata.get('mutation_totals', {}) + self.assertGreaterEqual(mutation_totals.get('assistant_delta_append', 0), 2) + self.assertEqual(mutation_totals.get('assistant_finalize', 0), 1) def test_agent_streams_tool_calls_and_reconstructs_arguments(self) -> None: responses = [ @@ -439,6 +447,15 @@ class AgentRuntimeTests(unittest.TestCase): self.assertEqual(result.usage.output_tokens, 6) assistant_messages = [message for message in result.transcript if message.get('role') == 'assistant'] self.assertTrue(any(message.get('tool_calls') for message in assistant_messages)) + assistant_with_tool = next( + message + for message in assistant_messages + if message.get('tool_calls') + ) + metadata = assistant_with_tool.get('metadata', {}) + mutation_totals = metadata.get('mutation_totals', {}) + self.assertGreaterEqual(mutation_totals.get('assistant_tool_call_delta', 0), 2) + self.assertEqual(mutation_totals.get('assistant_finalize', 0), 1) def test_transcript_entries_include_structured_blocks(self) -> None: responses = [ @@ -578,6 +595,10 @@ class AgentRuntimeTests(unittest.TestCase): if message.get('metadata', {}).get('kind') == 'compact_boundary' ] self.assertEqual(len(transcript_compact_messages), 1) + compact_metadata = transcript_compact_messages[0].get('metadata', {}) + self.assertEqual(compact_metadata.get('compaction_depth'), 1) + self.assertEqual(compact_metadata.get('nested_compaction_count'), 0) + self.assertIn('preserved_tail_ids', compact_metadata) def test_agent_enforces_total_token_budget(self) -> None: responses = [ @@ -1386,6 +1407,9 @@ class AgentRuntimeTests(unittest.TestCase): self.assertEqual(len(metadata.get('child_session_ids', [])), 2) self.assertEqual(len(metadata.get('child_results', [])), 2) self.assertIn('Delegated agent completed 2 sequential subtasks.', tool_messages[0].get('content', '')) + self.assertTrue(any(event.get('type') == 'delegate_group_result' for event in result.events)) + child_events = [event for event in result.events if event.get('type') == 'delegate_subtask_result'] + self.assertEqual(len(child_events), 2) second_child_request = recorded_payloads[2]['messages'] assert isinstance(second_child_request, list) self.assertTrue( @@ -1497,6 +1521,153 @@ class AgentRuntimeTests(unittest.TestCase): summary = '\n'.join(manager.summary_lines()) self.assertIn(f'group={group.group_id}', summary) + def test_agent_can_delegate_into_resumed_child_session(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Seed child result.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 5, 'completion_tokens': 2}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Delegating into a resumed child session.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'delegate_agent', + 'arguments': '{}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ], + 'usage': {'prompt_tokens': 8, 'completion_tokens': 3}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Resumed child result.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 6, 'completion_tokens': 2}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Parent completed after resumed child.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 7, 'completion_tokens': 2}, + }, + ] + recorded_payloads: list[dict[str, object]] = [] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + session_dir = workspace / '.port_sessions' / 'agent' + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads), + ): + seed_agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig( + cwd=workspace, + session_directory=session_dir, + ), + ) + seeded = seed_agent.run('Seed the delegated child') + resumed_child_id = seeded.session_id or '' + + delegate_arguments = json.dumps( + { + 'subtasks': [ + { + 'label': 'resume_child', + 'prompt': 'Continue the delegated child.', + 'resume_session_id': resumed_child_id, + 'max_turns': 2, + } + ], + 'max_turns': 2, + } + ) + responses[1]['choices'][0]['message']['tool_calls'][0]['function']['arguments'] = delegate_arguments + + parent_agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig( + cwd=workspace, + session_directory=session_dir, + ), + ) + result = parent_agent.run('Delegate into the resumed child') + + self.assertEqual(result.final_output, 'Parent completed after resumed child.') + tool_messages = [message for message in result.transcript if message.get('role') == 'tool'] + self.assertEqual(len(tool_messages), 1) + metadata = tool_messages[0].get('metadata', {}) + self.assertEqual(metadata.get('resumed_children'), 1) + child_results = metadata.get('child_results', []) + self.assertEqual(len(child_results), 1) + self.assertEqual(child_results[0].get('resume_used'), True) + self.assertEqual(child_results[0].get('resumed_from_session_id'), resumed_child_id) + self.assertTrue( + any( + event.get('type') == 'delegate_subtask_result' + and event.get('resume_used') + for event in result.events + ) + ) + resumed_child_messages = recorded_payloads[2]['messages'] + assert isinstance(resumed_child_messages, list) + resumed_contents = [ + message.get('content') + for message in resumed_child_messages + if isinstance(message, dict) + ] + self.assertIn('Seed the delegated child', resumed_contents) + self.assertIn('Seed child result.', resumed_contents) + self.assertIn('Continue the delegated child.', resumed_contents) + manager = parent_agent.agent_manager + assert manager is not None + child_records = [ + record + for record in manager.completed_records() + if record.parent_agent_id == parent_agent.managed_agent_id + ] + self.assertEqual(len(child_records), 1) + self.assertEqual(child_records[0].resumed_from_session_id, resumed_child_id) + summary = '\n'.join(manager.summary_lines()) + self.assertIn(f'resumed_from={resumed_child_id}', summary) + def test_agent_enforces_reasoning_token_budget(self) -> None: responses = [ { diff --git a/tests/test_query_engine_runtime.py b/tests/test_query_engine_runtime.py index 2cf531b..37d0a2b 100644 --- a/tests/test_query_engine_runtime.py +++ b/tests/test_query_engine_runtime.py @@ -10,7 +10,7 @@ from src.agent_runtime import LocalCodingAgent from src.agent_types import AgentRuntimeConfig, ModelConfig from src.openai_compat import OpenAICompatClient from src.plugin_runtime import PluginRuntime -from src.query_engine import QueryEnginePort +from src.query_engine import QueryEngineConfig, QueryEnginePort class FakeHTTPResponse: @@ -27,6 +27,27 @@ class FakeHTTPResponse: return None +class FakeStreamingHTTPResponse: + def __init__(self, payloads: list[dict[str, object]]) -> None: + self.lines: list[bytes] = [] + for payload in payloads: + chunk = f'data: {json.dumps(payload)}\n\n' + self.lines.extend(part.encode('utf-8') for part in chunk.splitlines(keepends=True)) + done_chunk = 'data: [DONE]\n\n' + self.lines.extend(part.encode('utf-8') for part in done_chunk.splitlines(keepends=True)) + + def readline(self) -> bytes: + if not self.lines: + return b'' + return self.lines.pop(0) + + def __enter__(self) -> 'FakeStreamingHTTPResponse': + return self + + def __exit__(self, exc_type, exc, tb) -> None: + return None + + def make_recording_urlopen_side_effect( responses: list[dict[str, object]], recorded_payloads: list[dict[str, object]], @@ -50,6 +71,17 @@ def make_urlopen_side_effect(responses: list[dict[str, object]]): return _fake_urlopen +def make_streaming_urlopen_side_effect( + responses: list[list[dict[str, object]]], +): + queued = [FakeStreamingHTTPResponse(payloads) for payloads in responses] + + def _fake_urlopen(request_obj, timeout=None): # noqa: ANN001 + return queued.pop(0) + + return _fake_urlopen + + class QueryEngineRuntimeTests(unittest.TestCase): def test_plugin_runtime_discovers_local_manifest(self) -> None: with tempfile.TemporaryDirectory() as tmp_dir: @@ -249,6 +281,112 @@ class QueryEngineRuntimeTests(unittest.TestCase): ) ) + def test_runtime_agent_executes_plugin_virtual_tool(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Calling the plugin virtual tool.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'demo_virtual', + 'arguments': '{"topic": "plugins"}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ], + 'usage': {'prompt_tokens': 8, 'completion_tokens': 3}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Plugin virtual tool completed.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 7, 'completion_tokens': 2}, + }, + ] + recorded_payloads: list[dict[str, object]] = [] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + plugin_dir = workspace / 'plugins' / 'demo' + plugin_dir.mkdir(parents=True) + (plugin_dir / 'plugin.json').write_text( + json.dumps( + { + 'name': 'demo-plugin', + 'virtualTools': [ + { + 'name': 'demo_virtual', + 'description': 'Return a rendered plugin response.', + 'responseTemplate': 'Plugin says {topic}', + 'parameters': { + 'type': 'object', + 'properties': {'topic': {'type': 'string'}}, + 'required': ['topic'], + }, + } + ], + 'toolHooks': { + 'demo_virtual': { + 'afterResult': 'Use the virtual tool result in the next reply.', + } + }, + } + ), + encoding='utf-8', + ) + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace), + ) + result = agent.run('Use the plugin virtual tool') + + self.assertEqual(result.final_output, 'Plugin virtual tool completed.') + self.assertTrue( + any(event.get('type') == 'plugin_virtual_tool_result' for event in result.events) + ) + tool_names = [ + item['function']['name'] + for item in recorded_payloads[0]['tools'] + if isinstance(item, dict) and isinstance(item.get('function'), dict) + ] + self.assertIn('demo_virtual', tool_names) + tool_messages = [message for message in result.transcript if message.get('role') == 'tool'] + self.assertEqual(len(tool_messages), 1) + metadata = tool_messages[0].get('metadata', {}) + self.assertEqual(metadata.get('action'), 'plugin_virtual_tool') + self.assertEqual(metadata.get('plugin_name'), 'demo-plugin') + self.assertEqual(metadata.get('virtual_tool'), 'demo_virtual') + second_messages = recorded_payloads[1]['messages'] + assert isinstance(second_messages, list) + self.assertTrue( + any( + isinstance(message, dict) + and 'Use the virtual tool result in the next reply.' in str(message.get('content', '')) + for message in second_messages + ) + ) + def test_runtime_agent_injects_plugin_tool_runtime_guidance(self) -> None: responses = [ { @@ -337,6 +475,93 @@ class QueryEngineRuntimeTests(unittest.TestCase): ) ) + def test_runtime_agent_supports_plugin_before_tool_guidance(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Reading through plugin guidance.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'read_file', + 'arguments': '{"path": "guide.txt"}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ], + 'usage': {'prompt_tokens': 8, 'completion_tokens': 3}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Plugin before-tool guidance consumed.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 7, 'completion_tokens': 2}, + }, + ] + recorded_payloads: list[dict[str, object]] = [] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + (workspace / 'guide.txt').write_text('plugin before tool\n', encoding='utf-8') + plugin_dir = workspace / 'plugins' / 'demo' + plugin_dir.mkdir(parents=True) + (plugin_dir / 'plugin.json').write_text( + json.dumps( + { + 'name': 'demo-plugin', + 'toolHooks': { + 'read_file': { + 'beforeTool': 'Validate the path before reading.', + } + }, + } + ), + encoding='utf-8', + ) + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_recording_urlopen_side_effect(responses, recorded_payloads), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace), + ) + result = agent.run('Read the file and continue') + + self.assertEqual(result.final_output, 'Plugin before-tool guidance consumed.') + self.assertTrue(any(event.get('type') == 'plugin_tool_preflight' for event in result.events)) + runtime_messages = [ + message for message in result.transcript + if message.get('metadata', {}).get('kind') == 'plugin_tool_runtime' + ] + self.assertEqual(len(runtime_messages), 1) + self.assertIn('Before tool: demo-plugin: Validate the path before reading.', runtime_messages[0].get('content', '')) + second_messages = recorded_payloads[1]['messages'] + assert isinstance(second_messages, list) + self.assertTrue( + any( + isinstance(message, dict) + and 'Before tool: demo-plugin: Validate the path before reading.' in str(message.get('content', '')) + for message in second_messages + ) + ) + def test_runtime_agent_blocks_tool_via_plugin_manifest(self) -> None: responses = [ { @@ -495,3 +720,317 @@ class QueryEngineRuntimeTests(unittest.TestCase): self.assertIn('## Runtime Message Kinds', summary) self.assertIn('- plugin_tool_runtime=1', summary) self.assertIn('- transcript_messages=', summary) + + def test_query_engine_runtime_stream_emits_runtime_summary_event(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Streaming summary ready.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 8, 'completion_tokens': 3}, + } + ] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_urlopen_side_effect(responses), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace), + ) + engine = QueryEnginePort.from_runtime_agent(agent) + events = list(engine.stream_submit_message('Summarize the repo')) + + runtime_summary_events = [ + event for event in events if event.get('type') == 'runtime_summary' + ] + self.assertEqual(len(runtime_summary_events), 1) + summary_event = runtime_summary_events[0] + self.assertIn('runtime_event_counts', summary_event) + self.assertIn('transcript_store_entries', summary_event) + self.assertIn('transcript_store_compactions', summary_event) + + def test_query_engine_compacts_transcript_store_with_summary_entry(self) -> None: + engine = QueryEnginePort.from_workspace() + engine.config = QueryEngineConfig(max_turns=6, compact_after_turns=2) + engine.submit_message('first prompt') + engine.submit_message('second prompt') + engine.submit_message('third prompt') + + replay = engine.replay_user_messages() + self.assertTrue(any('[transcript-compaction ' in entry for entry in replay)) + summary = engine.render_summary() + self.assertIn('## Transcript Store', summary) + self.assertIn('Transcript compactions:', summary) + + def test_query_engine_runtime_summary_tracks_mutation_counts(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Reading the large file first.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'read_file', + 'arguments': '{"path": "large.txt"}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ], + 'usage': {'prompt_tokens': 8, 'completion_tokens': 3}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Mutation summary completed.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 7, 'completion_tokens': 2}, + }, + ] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + (workspace / 'large.txt').write_text(('alpha beta gamma\n' * 400), encoding='utf-8') + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_urlopen_side_effect(responses), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig( + cwd=workspace, + auto_snip_threshold_tokens=120, + compact_preserve_messages=0, + ), + ) + engine = QueryEnginePort.from_runtime_agent(agent) + turn = engine.submit_message('Read the large file and summarize it') + summary = engine.render_summary() + + self.assertEqual(turn.output, 'Mutation summary completed.') + self.assertIn('## Runtime Mutations', summary) + self.assertIn('- snip_tombstone=1', summary) + self.assertIn('- tool_finalize_replace=1', summary) + self.assertIn('## Runtime Context Reduction', summary) + self.assertIn('- snipped_messages=1', summary) + + def test_query_engine_runtime_summary_tracks_compaction_lineage(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Read the large file first.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'read_file', + 'arguments': '{"path": "large.txt"}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ], + 'usage': {'prompt_tokens': 8, 'completion_tokens': 3}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Compaction lineage summary completed.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 7, 'completion_tokens': 2}, + }, + ] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + (workspace / 'large.txt').write_text(('alpha beta gamma\n' * 400), encoding='utf-8') + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_urlopen_side_effect(responses), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig( + cwd=workspace, + auto_compact_threshold_tokens=120, + compact_preserve_messages=0, + ), + ) + engine = QueryEnginePort.from_runtime_agent(agent) + turn = engine.submit_message('Read the large file and summarize it') + summary = engine.render_summary() + + self.assertEqual(turn.output, 'Compaction lineage summary completed.') + self.assertIn('## Runtime Context Reduction', summary) + self.assertIn('- compact_boundaries=1', summary) + self.assertIn('- compacted_lineages=', summary) + self.assertIn('## Runtime Lineage', summary) + self.assertIn('- seen_lineages=', summary) + self.assertIn('- compacted_lineages=', summary) + self.assertIn('- max_source_revision=', summary) + + def test_query_engine_runtime_summary_tracks_assistant_stream_mutations(self) -> None: + responses = [ + [ + {'choices': [{'delta': {'content': 'Streaming '}, 'finish_reason': None}]}, + {'choices': [{'delta': {'content': 'works.'}, 'finish_reason': None}]}, + { + 'choices': [{'delta': {}, 'finish_reason': 'stop'}], + 'usage': {'prompt_tokens': 14, 'completion_tokens': 5}, + }, + ] + ] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_streaming_urlopen_side_effect(responses), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig( + cwd=workspace, + stream_model_responses=True, + ), + ) + engine = QueryEnginePort.from_runtime_agent(agent) + turn = engine.submit_message('Say streaming works') + summary = engine.render_summary() + + self.assertEqual(turn.output, 'Streaming works.') + self.assertIn('## Runtime Mutations', summary) + self.assertIn('- assistant_delta_append=2', summary) + self.assertIn('- assistant_finalize=1', summary) + + def test_query_engine_runtime_summary_tracks_delegate_orchestration(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Delegating multiple subtasks.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'delegate_agent', + 'arguments': json.dumps( + { + 'subtasks': [ + {'label': 'scan', 'prompt': 'Scan the project.'}, + {'label': 'summarize', 'prompt': 'Summarize the project.'}, + ], + 'max_turns': 2, + } + ), + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ], + 'usage': {'prompt_tokens': 8, 'completion_tokens': 3}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Child scan result.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 5, 'completion_tokens': 2}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Child summary result.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 6, 'completion_tokens': 2}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Parent completed after multi-delegate.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 7, 'completion_tokens': 2}, + }, + ] + with tempfile.TemporaryDirectory() as tmp_dir: + workspace = Path(tmp_dir) + with patch( + 'src.openai_compat.request.urlopen', + side_effect=make_urlopen_side_effect(responses), + ): + agent = LocalCodingAgent( + model_config=ModelConfig( + model='Qwen/Qwen3-Coder-30B-A3B-Instruct', + base_url='http://127.0.0.1:8000/v1', + ), + runtime_config=AgentRuntimeConfig(cwd=workspace), + ) + engine = QueryEnginePort.from_runtime_agent(agent) + turn = engine.submit_message('Use multiple delegated subtasks') + summary = engine.render_summary() + + self.assertEqual(turn.output, 'Parent completed after multi-delegate.') + self.assertIn('## Runtime Orchestration', summary) + self.assertIn('- group_status:completed=1', summary) + self.assertIn('- child_stop:stop=2', summary)