From 5ad10c660c462bdeef454979e68bede448211d29 Mon Sep 17 00:00:00 2001 From: Abdelrahman Abdallah Date: Thu, 2 Apr 2026 22:19:32 +0200 Subject: [PATCH] add File-history snapshot ids and replay summaries for file edits --- PARITY_CHECKLIST.md | 7 +- src/agent_runtime.py | 83 ++++++++++++++++++++- src/agent_tools.py | 10 ++- tests/test_agent_runtime.py | 74 +++++++++++++++++++ tests/test_query_engine_runtime.py | 114 +++++++++++++++++++++++++++++ 5 files changed, 283 insertions(+), 5 deletions(-) diff --git a/PARITY_CHECKLIST.md b/PARITY_CHECKLIST.md index fd14493..dcb3372 100644 --- a/PARITY_CHECKLIST.md +++ b/PARITY_CHECKLIST.md @@ -24,12 +24,15 @@ Done: - [x] File history journaling for write/edit/shell tool actions - [x] Incremental `bash` tool-result streaming events - [x] Incremental tool-result streaming for read-only text tools +- [x] Incremental tool-result streaming across the current Python text tool surface - [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 +- [x] File-history snapshot ids and replay summaries for file edits +- [x] File-history result previews for shell and delegated-tool entries - [x] Truncated-response continuation flow for `finish_reason=length` - [x] Basic snipping of older tool/tool-call messages for context control - [x] Basic automatic compact-boundary insertion with preserved recent tail @@ -69,11 +72,11 @@ Done: Missing: -- [ ] Full partial tool-result streaming parity across the complete tool surface +- [ ] Full partial tool-result streaming parity across the complete upstream/npm tool surface - [ ] Full rich transcript mutation behavior like the npm runtime - [ ] Full reasoning budgets and task budgets parity - [ ] Full multi-agent orchestration parity beyond sequential grouped delegation and resumed-child flows -- [ ] Full file history snapshots and replay flows +- [ ] Full file history snapshots and replay flows beyond the current preview/id-based implementation - [ ] 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 diff --git a/src/agent_runtime.py b/src/agent_runtime.py index 6256db4..08366b3 100644 --- a/src/agent_runtime.py +++ b/src/agent_runtime.py @@ -1159,14 +1159,35 @@ class LocalCodingAgent: and tool_result.metadata.get('action') != 'delegate_agent' ): return None - return { + metadata = dict(tool_result.metadata) + entry: dict[str, object] = { 'timestamp': datetime.now(timezone.utc).isoformat(), 'turn_index': turn_index, 'tool_call_id': tool_call.id, 'tool_name': tool_call.name, 'ok': tool_result.ok, - **dict(tool_result.metadata), + 'history_entry_id': f'{turn_index}:{tool_call.id}:{tool_call.name}', + 'result_preview': self._preview_text(tool_result.content, 220), + **metadata, } + action = metadata.get('action') + path = metadata.get('path') + if isinstance(path, str) and path: + entry['history_kind'] = 'file_change' + entry['changed_paths'] = [path] + before_sha256 = metadata.get('before_sha256') + if isinstance(before_sha256, str) and before_sha256: + entry['before_snapshot_id'] = f'{path}:{before_sha256[:12]}' + after_sha256 = metadata.get('after_sha256') + if isinstance(after_sha256, str) and after_sha256: + entry['after_snapshot_id'] = f'{path}:{after_sha256[:12]}' + elif isinstance(metadata.get('command'), str): + entry['history_kind'] = 'shell' + elif action == 'delegate_agent': + entry['history_kind'] = 'delegation' + else: + entry['history_kind'] = 'tool' + return entry def _compact_prefix_count(self, session: AgentSessionState) -> int: prefix_count = 0 @@ -1711,6 +1732,24 @@ class LocalCodingAgent: if not file_history: return replay_count = len(file_history) + unique_paths = sorted( + { + path + for entry in file_history + for path in ( + entry.get('changed_paths') + if isinstance(entry.get('changed_paths'), list) + else ([entry.get('path')] if isinstance(entry.get('path'), str) else []) + ) + if isinstance(path, str) and path + } + ) + snapshot_count = sum( + 1 + for entry in file_history + for key in ('before_snapshot_id', 'after_snapshot_id') + if isinstance(entry.get(key), str) and entry.get(key) + ) for message in reversed(session.messages): if message.metadata.get('kind') != 'file_history_replay': continue @@ -1722,6 +1761,8 @@ class LocalCodingAgent: metadata={ 'kind': 'file_history_replay', 'file_history_count': replay_count, + 'file_history_unique_paths': len(unique_paths), + 'file_history_snapshot_count': snapshot_count, }, message_id=f'file_history_replay_{replay_count}', ) @@ -1730,16 +1771,45 @@ class LocalCodingAgent: self, file_history: tuple[dict[str, object], ...], ) -> str: + unique_paths = sorted( + { + path + for entry in file_history + for path in ( + entry.get('changed_paths') + if isinstance(entry.get('changed_paths'), list) + else ([entry.get('path')] if isinstance(entry.get('path'), str) else []) + ) + if isinstance(path, str) and path + } + ) + snapshot_count = sum( + 1 + for entry in file_history + for key in ('before_snapshot_id', 'after_snapshot_id') + if isinstance(entry.get(key), str) and entry.get(key) + ) lines = [ '', 'Recent file history from this saved session:', + f'- History entries: {len(file_history)}', + f'- Unique changed paths: {len(unique_paths)}', + f'- Snapshot ids: {snapshot_count}', ] + if unique_paths: + preview_paths = ', '.join(unique_paths[:4]) + if len(unique_paths) > 4: + preview_paths += f', ... (+{len(unique_paths) - 4} more)' + lines.append(f'- Changed path preview: {preview_paths}') for entry in file_history[-10:]: action = str(entry.get('action', entry.get('tool_name', 'tool'))) turn = entry.get('turn_index') path = entry.get('path') command = entry.get('command') details = [f'action={action}'] + history_entry_id = entry.get('history_entry_id') + if isinstance(history_entry_id, str) and history_entry_id: + details.append(f'entry_id={history_entry_id}') if turn is not None: details.append(f'turn={turn}') if path: @@ -1750,12 +1820,21 @@ class LocalCodingAgent: if isinstance(child_session_ids, list) and child_session_ids: details.append(f'child_sessions={len(child_session_ids)}') lines.append(f"- {'; '.join(details)}") + before_snapshot_id = entry.get('before_snapshot_id') + if isinstance(before_snapshot_id, str) and before_snapshot_id: + lines.append(f' before_snapshot: {before_snapshot_id}') + after_snapshot_id = entry.get('after_snapshot_id') + if isinstance(after_snapshot_id, str) and after_snapshot_id: + lines.append(f' after_snapshot: {after_snapshot_id}') before_preview = entry.get('before_preview') if isinstance(before_preview, str) and before_preview: lines.append(f' before: {before_preview}') after_preview = entry.get('after_preview') if isinstance(after_preview, str) and after_preview: lines.append(f' after: {after_preview}') + result_preview = entry.get('result_preview') + if isinstance(result_preview, str) and result_preview: + lines.append(f' result: {result_preview}') if len(file_history) > 10: lines.append(f'- ... plus {len(file_history) - 10} older file-history entries') lines.extend( diff --git a/src/agent_tools.py b/src/agent_tools.py index 96990fd..731be3f 100644 --- a/src/agent_tools.py +++ b/src/agent_tools.py @@ -121,7 +121,7 @@ def execute_tool_streaming( return result = tool.execute(arguments, context) - if name in {'list_dir', 'read_file', 'glob_search', 'grep_search'} and result.ok: + if result.ok and result.content and name != 'delegate_agent': yield from _stream_static_text_result(result) return yield ToolStreamUpdate(kind='result', result=result) @@ -541,6 +541,9 @@ def _run_bash(arguments: dict[str, Any], context: ToolExecutionContext) -> str: 'action': 'bash', 'command': command, 'exit_code': completed.returncode, + 'stdout_preview': _snapshot_text(stdout), + 'stderr_preview': _snapshot_text(stderr), + 'output_preview': _snapshot_text('\n'.join(payload).strip()), }, ) @@ -634,6 +637,8 @@ def _stream_bash( 'command': command, 'exit_code': exit_code, 'timed_out': True, + 'stdout_preview': _snapshot_text(''.join(stdout_chunks)), + 'stderr_preview': _snapshot_text(''.join(stderr_chunks)), }, ), ) @@ -659,6 +664,9 @@ def _stream_bash( 'command': command, 'exit_code': exit_code, 'streamed': True, + 'stdout_preview': _snapshot_text(stdout), + 'stderr_preview': _snapshot_text(stderr), + 'output_preview': _snapshot_text('\n'.join(payload).strip()), }, ), ) diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py index 66e79f4..87ddf7b 100644 --- a/tests/test_agent_runtime.py +++ b/tests/test_agent_runtime.py @@ -749,7 +749,76 @@ class AgentRuntimeTests(unittest.TestCase): stored = load_agent_session(result.session_id or '', directory=session_dir) self.assertEqual(len(result.file_history), 1) self.assertEqual(result.file_history[0]['path'], 'out.txt') + self.assertEqual(result.file_history[0]['history_kind'], 'file_change') + self.assertIn('history_entry_id', result.file_history[0]) + self.assertIn('after_snapshot_id', result.file_history[0]) self.assertEqual(stored.file_history[0]['action'], 'write_file') + self.assertEqual(stored.file_history[0]['after_snapshot_id'], result.file_history[0]['after_snapshot_id']) + + def test_agent_streams_write_file_tool_output(self) -> None: + responses = [ + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Creating the file.', + 'tool_calls': [ + { + 'id': 'call_1', + 'type': 'function', + 'function': { + 'name': 'write_file', + 'arguments': '{"path": "out.txt", "content": "hello"}', + }, + } + ], + }, + 'finish_reason': 'tool_calls', + } + ], + 'usage': {'prompt_tokens': 4, 'completion_tokens': 3}, + }, + { + 'choices': [ + { + 'message': { + 'role': 'assistant', + 'content': 'Write streamed.', + }, + 'finish_reason': 'stop', + } + ], + 'usage': {'prompt_tokens': 5, '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, + permissions=AgentPermissions(allow_file_write=True), + ), + ) + result = agent.run('Create out.txt') + self.assertEqual(result.final_output, 'Write streamed.') + tool_delta_events = [ + event for event in result.events + if event.get('type') == 'tool_delta' and event.get('tool_name') == 'write_file' + ] + self.assertGreaterEqual(len(tool_delta_events), 1) + tool_message = next( + message for message in result.transcript + if message.get('role') == 'tool' + ) + metadata = tool_message.get('metadata', {}) + self.assertEqual(metadata.get('streamed'), True) + self.assertEqual(metadata.get('action'), 'write_file') def test_agent_streams_bash_tool_output_and_mutates_tool_transcript(self) -> None: responses = [ @@ -1142,8 +1211,13 @@ class AgentRuntimeTests(unittest.TestCase): ] self.assertEqual(len(replay_messages), 1) replay_content = replay_messages[0]['content'] + self.assertIn('Unique changed paths: 1', replay_content) + self.assertIn('Snapshot ids: 2', replay_content) + self.assertIn('before_snapshot:', replay_content) + self.assertIn('after_snapshot:', replay_content) self.assertIn('before: hello world', replay_content) self.assertIn('after: hello mars', replay_content) + self.assertIn('result:', replay_content) def test_resume_injects_compaction_replay_reminder(self) -> None: responses = [ diff --git a/tests/test_query_engine_runtime.py b/tests/test_query_engine_runtime.py index 37d0a2b..e535037 100644 --- a/tests/test_query_engine_runtime.py +++ b/tests/test_query_engine_runtime.py @@ -1034,3 +1034,117 @@ class QueryEngineRuntimeTests(unittest.TestCase): self.assertIn('## Runtime Orchestration', summary) self.assertIn('- group_status:completed=1', summary) self.assertIn('- child_stop:stop=2', summary) + + def test_query_engine_runtime_summary_tracks_resumed_delegate_children(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 resumed child.', + '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}, + }, + ] + 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_urlopen_side_effect(responses), + ): + 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 + + 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, + ), + ) + engine = QueryEnginePort.from_runtime_agent(agent) + turn = engine.submit_message('Delegate into resumed child') + summary = engine.render_summary() + + self.assertEqual(turn.output, 'Parent completed after resumed child.') + self.assertIn('## Runtime Orchestration', summary) + self.assertIn('- resumed_children=1', summary)