import asyncio import json import sys import types import pytest def _install_acp_stubs(): acp_module = types.ModuleType("acp") acp_module.PROTOCOL_VERSION = 1 acp_module.Client = object acp_module.RequestError = Exception acp_module.connect_to_agent = object acp_module.image_block = lambda data, mime_type: {"data": data, "mimeType": mime_type} acp_module.text_block = lambda text: text sys.modules.setdefault("acp", acp_module) core_module = types.ModuleType("acp.core") core_module.ClientSideConnection = object sys.modules.setdefault("acp.core", core_module) schema_module = types.ModuleType("acp.schema") for name in ( "AgentMessageChunk", "AgentPlanUpdate", "AgentThoughtChunk", "AllowedOutcome", "AvailableCommandsUpdate", "ClientCapabilities", "CurrentModeUpdate", "DeniedOutcome", "EmbeddedResourceContentBlock", "FileSystemCapabilities", "PermissionOption", "ReadTextFileResponse", "RequestPermissionResponse", "ResourceContentBlock", "SessionInfoUpdate", "TextContentBlock", "ToolCall", "ToolCallProgress", "ToolCallStart", "UsageUpdate", "UserMessageChunk", "WriteTextFileResponse", ): setattr(schema_module, name, type(name, (), {})) sys.modules.setdefault("acp.schema", schema_module) _install_acp_stubs() from acp.schema import AgentMessageChunk, AgentThoughtChunk, ToolCallProgress, ToolCallStart # noqa: E402 from investibook.core import agent_runtime as runtime_mod # noqa: E402 from investibook.core.agent_runtime import AgentSession, AgentEventBridge # noqa: E402 CODE_INVESTIGATOR_ENVELOPE_MARKER = "" # V2 Task results are raw JSON. # noqa: E402 def _agent_thought_chunk(**kwargs): try: return AgentThoughtChunk( sessionUpdate="agent_thought_chunk", content=kwargs.get("content"), _meta=kwargs.get("field_meta"), ) except TypeError: update = AgentThoughtChunk() update.content = kwargs.get("content") update.field_meta = kwargs.get("field_meta") return update def _agent_message_chunk(**kwargs): try: return AgentMessageChunk( sessionUpdate="agent_message_chunk", content=kwargs.get("content"), _meta=kwargs.get("field_meta"), ) except TypeError: update = AgentMessageChunk() update.content = kwargs.get("content") update.field_meta = kwargs.get("field_meta") return update def _tool_call_start(**kwargs): try: return ToolCallStart( toolCallId=kwargs.get("tool_call_id", ""), sessionUpdate="tool_call", title=kwargs.get("title"), rawInput=kwargs.get("raw_input"), _meta=kwargs.get("field_meta"), ) except TypeError: update = ToolCallStart() update.tool_call_id = kwargs.get("tool_call_id", "") update.title = kwargs.get("title") update.raw_input = kwargs.get("raw_input") update.field_meta = kwargs.get("field_meta") return update def _tool_call_progress(**kwargs): try: return ToolCallProgress( toolCallId=kwargs["tool_call_id"], sessionUpdate="tool_call_update", title=kwargs.get("title"), rawInput=kwargs.get("raw_input"), rawOutput=kwargs.get("raw_output"), status=kwargs.get("status"), _meta=kwargs.get("field_meta"), ) except TypeError: update = ToolCallProgress() update.tool_call_id = kwargs["tool_call_id"] update.title = kwargs.get("title") update.raw_input = kwargs.get("raw_input") update.raw_output = kwargs.get("raw_output") update.status = kwargs.get("status") update.field_meta = kwargs.get("field_meta") return update def test_acp_session_exposes_workspace_id(tmp_path): workspace_dir = tmp_path / "ws-123" workspace_dir.mkdir() session = AgentSession("frontend", "opencode", workspace_dir=workspace_dir) assert session.workspace_id == "ws-123" def test_session_update_does_not_auto_append_legacy_free_form_output(monkeypatch): client = AgentEventBridge(workspace_id="ws-auto") client.start_assistant_message("assistant-1") update = _tool_call_progress( tool_call_id="tool-1", title="QPS query", raw_input={"query": "sum(rate(tidb_server_query_total[1m]))"}, raw_output={ "output": ( '{"status":"success","data":{"resultType":"matrix","result":' '[{"metric":{"instance":"db-tidb-0:10080"},"values":[["1","10"],["2","20"]]}]}}' ), "metadata": {"exit": 0}, }, status="completed", ) asyncio.run(client.session_update("session-1", update)) assert client._msg_id == "assistant-1" def test_session_update_ignores_structured_workspace_evidence_envelope(monkeypatch): client = AgentEventBridge(workspace_id="ws-auto") client.start_assistant_message("assistant-structured") update = _tool_call_progress( tool_call_id="tool-1", title="clinic-cli metrics", raw_input={"query": "sum(rate(tidb_server_query_total[1m]))"}, raw_output={ "workspace_evidence": { "kind": "metric", "cluster_id": "cluster-2", "source": "clinic-cli", "payload": { "metrics": [ { "name": "TiDB QPS", "promql": "sum(rate(tidb_server_query_total[1m]))", } ] }, }, "metadata": {"exit": 0}, }, status="completed", ) asyncio.run(client.session_update("session-1", update)) assert client._msg_id == "assistant-structured" def test_idless_failed_tool_progress_updates_existing_running_part(): client = AgentEventBridge(workspace_id="ws-refresh") client.start_assistant_message("assistant-refresh") start = _tool_call_start( title="bash", raw_input={"description": "Inspect backend logs"}, ) asyncio.run(client.session_update("session-1", start)) progress = _tool_call_progress( tool_call_id="", title="bash", raw_input={"command": "exit 1"}, raw_output={"output": "boom", "metadata": {"exit": 1}}, status="completed", ) asyncio.run(client.session_update("session-1", progress)) parts = [client._msg_parts[pid] for pid in client._msg_parts_order] tool_parts = [part for part in parts if part.get("type") == "tool"] assert len(tool_parts) == 1 assert tool_parts[0]["state"]["status"] == "failed" assert tool_parts[0]["state"]["output"]["metadata"]["exit"] == 1 def test_generic_tool_start_backfills_description_from_recent_thought(): client = AgentEventBridge(workspace_id="ws-enrich") client.start_assistant_message("assistant-enrich") thought = _agent_thought_chunk(content="我识别到这是 TiDB 集群性能诊断请求") asyncio.run(client.session_update("session-1", thought)) # A generic bash start with no description should be backfilled. start = _tool_call_start(title="bash", raw_input={"command": "ls"}) asyncio.run(client.session_update("session-1", start)) tool_parts = [p for p in client._msg_parts.values() if p.get("type") == "tool"] assert len(tool_parts) == 1 assert tool_parts[0]["state"]["input"].get("description") == "我识别到这是 TiDB 集群性能诊断请求" def test_concrete_tool_start_is_not_polluted_with_recent_thought(): client = AgentEventBridge(workspace_id="ws-enrich2") client.start_assistant_message("assistant-enrich2") thought = _agent_thought_chunk(content="我识别到这是 TiDB 集群性能诊断请求") asyncio.run(client.session_update("session-1", thought)) # A read whose title is the file path must NOT have a description injected. start = _tool_call_start( title="/ws/SKILL_CATALOG.md", raw_input={"filePath": "/ws/SKILL_CATALOG.md", "offset": 1, "limit": 2000}, ) asyncio.run(client.session_update("session-1", start)) tool_parts = [p for p in client._msg_parts.values() if p.get("type") == "tool"] assert len(tool_parts) == 1 assert "description" not in tool_parts[0]["state"]["input"] def test_tool_subagent_meta_is_emitted_and_persisted(): client = AgentEventBridge(workspace_id="ws-subagent") client.start_assistant_message("assistant-subagent") meta = { "opencode/subagent": { "parentSessionId": "parent-session", "parentTaskCallId": "task-call", "subagentSessionId": "child-session", "subagentName": "evidence-agent", } } start = _tool_call_start( tool_call_id="child-bash", title="bash", raw_input={"command": "echo hi"}, field_meta=meta, ) asyncio.run(client.session_update("session-1", start)) start_event = client.queue.get_nowait() progress = _tool_call_progress( tool_call_id="child-bash", title="bash", raw_input={"command": "echo hi"}, raw_output={"output": "hi\n", "metadata": {"exit": 0}}, status="completed", field_meta=meta, ) asyncio.run(client.session_update("session-1", progress)) progress_event = client.queue.get_nowait() part = client._msg_parts["tool-child-bash"] assert start_event["meta"] == meta assert progress_event["meta"] == meta assert part["state"]["meta"] == meta def test_code_investigator_artifact_only_accepts_registered_task(monkeypatch, tmp_path): client = AgentEventBridge(workspace_id="ws-code") client.start_assistant_message("assistant-code") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) delivered: dict[str, object] = {} def persist(**kwargs): delivered.update(kwargs) output = kwargs["workspace_root"] / "evidences" / "agent-runs" / "code-investigator.json" output.parent.mkdir(parents=True, exist_ok=True) output.write_text("{}") return output monkeypatch.setattr( "investibook.core.code_context.persist_code_investigator_envelope", persist ) artifact = CODE_INVESTIGATOR_ENVELOPE_MARKER + "\n{}" # A child bash completion that happens to contain a marker is never a task result. child = _tool_call_progress( tool_call_id="child-bash", title="bash", raw_output={"output": artifact}, status="completed", ) asyncio.run(client.session_update("session-1", child)) assert not delivered authorized_sha = "a" * 40 client.authorize_code_diagnosis_launch({"repo": "pingcap/tidb", "commit_sha": authorized_sha, "run_id": "turn-test"}) start = _tool_call_start( tool_call_id="task-1", title="TopSQL Others eviction source analysis", raw_input={}, ) task_input = { "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{authorized_sha}","focus_terms":["others"]}}\n' "inspect code" ), } asyncio.run(client.session_update("session-1", start)) task_running = _tool_call_progress( tool_call_id="task-1", title="TopSQL Others eviction source analysis", raw_input=task_input, raw_output=None, status="running", ) asyncio.run(client.session_update("session-1", task_running)) spoofed_submission = _tool_call_progress( tool_call_id="spoofed-submit-1", title="bash", raw_input={"artifact_json": artifact.strip()}, raw_output={"output": "not a controlled artifact submission"}, status="completed", field_meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, ) asyncio.run(client.session_update("session-1", spoofed_submission)) assert not delivered submission = _tool_call_progress( tool_call_id="submit-1", title="submit_code_artifact", raw_input={"artifact_json": artifact.strip()}, raw_output={"output": "artifact submission received"}, status="completed", field_meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, ) asyncio.run(client.session_update("session-1", submission)) completion = _tool_call_progress( tool_call_id="task-1", title="TopSQL Others eviction source analysis", raw_input=task_input, raw_output={ "output": "Source artifact submitted; runtime causality remains unverified." }, status="completed", ) asyncio.run(client.session_update("session-1", completion)) assert delivered["expected_task_id"] == "task-1" assert delivered["expected_session_id"] == "session-1" assert delivered["expected_commit_sha"] == authorized_sha assert delivered["expected_request_id"].startswith("code-") assert delivered["envelope"] == artifact.strip() events = [] while not client.queue.empty(): events.append(client.queue.get_nowait()) assert events[0]["output"]["output"] == artifact assert events[-1]["output"] == { "status": "artifact_ready", "artifact": "evidences/agent-runs/code-investigator.json", "schema_version": "1", } assert client.code_diagnosis_artifact_ready is True def test_code_diagnosis_hides_text_and_thought_until_artifact_is_ready(): client = AgentEventBridge(workspace_id="ws-code-output-gate") client.authorize_code_diagnosis_launch( {"repo": "pingcap/tidb", "commit_sha": "a" * 40, "run_id": "turn-test"} ) asyncio.run( client.session_update( "session-1", _agent_message_chunk(content="Unverified source conclusion") ) ) asyncio.run( client.session_update( "session-1", _agent_thought_chunk(content="Unverified reasoning") ) ) assert client.queue.empty() client._code_diagnosis_artifact_ready = True asyncio.run( client.session_update( "session-1", _agent_message_chunk(content="Artifact-backed conclusion") ) ) event = client.queue.get_nowait() assert event == {"type": "text", "text": "Artifact-backed conclusion"} def test_old_code_task_cannot_satisfy_a_new_code_diagnosis_launch(tmp_path): client = AgentEventBridge(workspace_id="ws-code-task-binding") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) sha = "a" * 40 launch = {"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-old"} client.authorize_code_diagnosis_launch(launch) client._register_code_investigator_task( tool_id="task-old", title="Task", raw_input={ "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["others"]}}\n' "inspect code" ), }, session_id="session-1", ) client._code_investigator_tasks["task-old"]["artifact_path"] = "evidences/agent-runs/code-investigator.json" client.authorize_code_diagnosis_launch( {"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-new"} ) result = asyncio.run( client._complete_code_investigator_task( tool_id="task-old", session_id="session-1", output={} ) ) assert result is not None assert result["status"] == "artifact_ready" assert client.code_diagnosis_artifact_ready is False def test_code_investigator_submission_requires_code_child_metadata(monkeypatch, tmp_path): client = AgentEventBridge(workspace_id="ws-code-submission-auth") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) persisted = False def persist(**kwargs): nonlocal persisted persisted = True raise AssertionError(f"unexpected persistence: {kwargs}") monkeypatch.setattr( "investibook.core.code_context.persist_code_investigator_envelope", persist ) sha = "a" * 40 client.authorize_code_diagnosis_launch({"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-test"}) client._register_code_investigator_task( tool_id="task-1", title="Task", raw_input={ "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["others"]}}\n' "inspect code" ), }, session_id="session-1", ) result = asyncio.run( client._submit_code_investigator_artifact( raw_input={"artifact_json": "{}"}, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "evidence-agent", } }, session_id="session-1", ) ) assert result == { "status": "artifact_submission_failed", "reason": "invalid_submission_context", } assert persisted is False def test_code_investigator_rejects_duplicate_artifact_submission(monkeypatch, tmp_path): client = AgentEventBridge(workspace_id="ws-code-submission-duplicate") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) persisted = 0 def persist(**kwargs): nonlocal persisted persisted += 1 output = kwargs["workspace_root"] / "evidences" / "agent-runs" / "code-investigator.json" output.parent.mkdir(parents=True, exist_ok=True) output.write_text("{}") return output monkeypatch.setattr( "investibook.core.code_context.persist_code_investigator_envelope", persist ) sha = "a" * 40 client.authorize_code_diagnosis_launch({"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-test"}) client._register_code_investigator_task( tool_id="task-1", title="Task", raw_input={ "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["others"]}}\n' "inspect code" ), }, session_id="session-1", ) metadata = { "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } } first = asyncio.run( client._submit_code_investigator_artifact( raw_input={"artifact_json": "{}"}, meta=metadata, session_id="session-1" ) ) second = asyncio.run( client._submit_code_investigator_artifact( raw_input={"artifact_json": "{}"}, meta=metadata, session_id="session-1" ) ) assert first["status"] == "artifact_submitted" assert second == { "status": "artifact_submission_failed", "reason": "artifact_already_submitted", } assert persisted == 1 def test_code_investigator_rejects_unextractable_artifact_payload(tmp_path): client = AgentEventBridge(workspace_id="ws-code-invalid-payload") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) sha = "a" * 40 client.authorize_code_diagnosis_launch( {"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-test"} ) client._register_code_investigator_task( tool_id="task-1", title="Task", raw_input={ "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["others"]}}\n' "inspect code" ), }, session_id="session-1", ) result = asyncio.run( client._submit_code_investigator_artifact( raw_input={"artifact": ["not", "an", "object"]}, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, session_id="session-1", ) ) assert result == { "status": "artifact_submission_failed", "reason": "invalid_artifact_payload", } def test_code_investigator_marks_artifact_ready_when_parent_task_is_failed(monkeypatch, tmp_path): client = AgentEventBridge(workspace_id="ws-code-out-of-order") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) def persist(**kwargs): output = kwargs["workspace_root"] / "evidences" / "agent-runs" / "code-investigator.json" output.parent.mkdir(parents=True, exist_ok=True) output.write_text("{}") return output monkeypatch.setattr( "investibook.core.code_context.persist_code_investigator_envelope", persist ) sha = "a" * 40 client.authorize_code_diagnosis_launch( {"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-test"} ) client._register_code_investigator_task( tool_id="task-1", title="Task", raw_input={ "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["others"]}}\n' "inspect code" ), }, session_id="session-1", ) early_completion = asyncio.run( client._complete_code_investigator_task( tool_id="task-1", session_id="session-1", output={"output": "done"} ) ) submission = asyncio.run( client._submit_code_investigator_artifact( raw_input={"artifact": {}}, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, session_id="session-1", ) ) assert early_completion == { "status": "artifact_delivery_failed", "reason": "missing_code_artifact_submission", } assert submission["status"] == "artifact_submitted" assert client.code_diagnosis_artifact_ready is True def test_code_investigator_artifact_requires_authorized_task_sha(monkeypatch, tmp_path): client = AgentEventBridge(workspace_id="ws-code-missing-sha") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) persisted = False def persist(**kwargs): nonlocal persisted persisted = True raise AssertionError(f"unexpected artifact persistence: {kwargs}") monkeypatch.setattr( "investibook.core.code_context.persist_code_investigator_envelope", persist ) client._register_code_investigator_task( tool_id="task-no-sha", title="Task", raw_input={"subagent_type": "code-investigator", "prompt": "inspect code"}, session_id="session-1", ) result = asyncio.run( client._complete_code_investigator_task( tool_id="task-no-sha", session_id="session-1", output={"output": CODE_INVESTIGATOR_ENVELOPE_MARKER + "\n{}"}, ) ) assert result == { "status": "artifact_delivery_failed", "reason": "invalid_code_context_task_header", } assert persisted is False def test_code_investigator_task_upgrade_replaces_null_request_id(monkeypatch, tmp_path): """A later full Task input must repair sparse registration correlation.""" client = AgentEventBridge(workspace_id="ws-code-request-id") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) delivered: dict[str, object] = {} def persist(**kwargs): delivered.update(kwargs) output = kwargs["workspace_root"] / "evidences" / "agent-runs" / "code-investigator.json" output.parent.mkdir(parents=True, exist_ok=True) output.write_text("{}") return output monkeypatch.setattr( "investibook.core.code_context.persist_code_investigator_envelope", persist ) sha = "a" * 40 client.authorize_code_diagnosis_launch({"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-test"}) client._code_investigator_tasks["task-1"] = { "session_id": "session-1", "workspace_id": "ws-code-request-id", "expected_commit_sha": "", "expected_focus_terms": "[]", "request_id": None, } client._register_code_investigator_task( tool_id="task-1", title="Task", raw_input={ "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["others"]}}\n' "inspect code" ), }, session_id="session-1", ) submission = asyncio.run( client._submit_code_investigator_artifact( raw_input={"artifact_json": "{}"}, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, session_id="session-1", ) ) result = asyncio.run( client._complete_code_investigator_task( tool_id="task-1", session_id="session-1", output={"output": "handoff"} ) ) assert submission["status"] == "artifact_submitted" assert result["status"] == "artifact_ready" assert isinstance(delivered["expected_request_id"], str) assert delivered["expected_request_id"].startswith("code-") def test_code_investigator_artifact_reports_validator_failure(monkeypatch, tmp_path): client = AgentEventBridge(workspace_id="ws-code-invalid-envelope") async def persist_snapshot(): return None client._agent_session = types.SimpleNamespace( workspace_dir=tmp_path, persist_assistant_message_snapshot=persist_snapshot, uses_sandbox=False, ) def persist(**kwargs): raise ValueError("artifact has unsupported fields: hypothesis_assessment") monkeypatch.setattr( "investibook.core.code_context.persist_code_investigator_envelope", persist ) sha = "a" * 40 client.authorize_code_diagnosis_launch({"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-test"}) client._register_code_investigator_task( tool_id="task-invalid-envelope", title="Task", raw_input={ "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["hypothesis"]}}\n' "inspect code" ), }, session_id="session-1", ) result = asyncio.run( client._submit_code_investigator_artifact( raw_input={"artifact_json": "{}"}, meta={ "opencode/subagent": { "parentTaskCallId": "task-invalid-envelope", "subagentName": "code-investigator", } }, session_id="session-1", ) ) assert result == { "status": "artifact_submission_failed", "reason": "invalid_envelope", "detail": "artifact has unsupported fields: hypothesis_assessment", } def test_structured_code_artifact_submission_escapes_special_characters() -> None: from investibook.core.agent_runtime import _code_artifact_submission submission = _code_artifact_submission( raw_input={ "artifact": { "question": '回答"有哪些系统表是内存表"', "notes": "line one\nline two\\path", } }, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, ) assert submission is not None parent_task_id, envelope = submission assert parent_task_id == "task-1" assert json.loads(envelope) == { "question": '回答"有哪些系统表是内存表"', "notes": "line one\nline two\\path", } def test_serialized_structured_code_artifact_submission_is_accepted() -> None: from investibook.core.agent_runtime import _code_artifact_submission artifact = { "question": '回答"有哪些系统表是内存表', "notes": "line one\nline two\\path", } submission = _code_artifact_submission( raw_input={"artifact": json.dumps(artifact, ensure_ascii=False)}, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, ) assert submission is not None parent_task_id, envelope = submission assert parent_task_id == "task-1" assert json.loads(envelope) == artifact def test_serialized_code_artifact_repairs_unescaped_model_quotes() -> None: from investibook.core.agent_runtime import _code_artifact_submission malformed = '{"question":"有两个不同的"慢日志文件"概念","notes":"line one\nline two"}' submission = _code_artifact_submission( raw_input={"artifact": malformed}, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, ) assert submission is not None parent_task_id, envelope = submission assert parent_task_id == "task-1" assert json.loads(envelope) == { "question": '有两个不同的"慢日志文件"概念', "notes": "line one\nline two", } def test_legacy_artifact_json_repairs_unescaped_model_quotes() -> None: from investibook.core.agent_runtime import _code_artifact_submission malformed = '{"question":"有两个不同的"慢日志文件"概念"}' submission = _code_artifact_submission( raw_input={"artifact_json": malformed}, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, ) assert submission is not None _, envelope = submission assert json.loads(envelope) == {"question": '有两个不同的"慢日志文件"概念'} def test_serialized_code_artifact_keeps_structural_json_errors_invalid() -> None: from investibook.core.agent_runtime import _code_artifact_submission malformed = '{"question":"missing separator" "notes":"still invalid"}' submission = _code_artifact_submission( raw_input={"artifact": malformed}, meta={ "opencode/subagent": { "parentTaskCallId": "task-1", "subagentName": "code-investigator", } }, ) assert submission is not None _, envelope = submission with pytest.raises(json.JSONDecodeError): json.loads(envelope) def test_code_investigator_task_header_requires_exact_lowercase_sha() -> None: from investibook.core.agent_runtime import _code_investigator_task_request def task_request( commit_sha: str, focus_terms: str = '["others"]', repo: str = "pingcap/tidb", ) -> dict[str, object]: return { "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"{repo}","commit_sha":"{commit_sha}","focus_terms":{focus_terms}}}\n' "inspect code" ), } sha = "a" * 40 assert _code_investigator_task_request(task_request(sha)) == { "repo": "pingcap/tidb", "commit_sha": sha, "focus_terms": ["others"], "mode": "initial", } assert _code_investigator_task_request(task_request(sha, repo="pingcap/tikv")) == { "repo": "pingcap/tikv", "commit_sha": sha, "focus_terms": ["others"], "mode": "initial", } assert _code_investigator_task_request(task_request(sha.upper())) is None assert _code_investigator_task_request(task_request(sha + " ")) is None assert _code_investigator_task_request(task_request(sha, "[]")) is None def test_code_investigator_task_header_preserves_focus_terms() -> None: from investibook.core.agent_runtime import _code_investigator_task_request sha = "a" * 40 request = { "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["others"]}}\n' "Explain others." ), } assert _code_investigator_task_request(request)["focus_terms"] == ["others"] assert _code_investigator_task_request(json.dumps(request))["commit_sha"] == sha def test_code_investigator_follow_up_header_requires_lineage() -> None: from investibook.core.agent_runtime import _code_investigator_task_request sha = "a" * 40 parent = "code-" + "1" * 32 request = { "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_FOLLOW_UP_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["partition state"],' f'"parent_request_id":"{parent}","follow_up_reason":"User asked to inspect the state machine."}}\n' "Inspect only the unresolved state transition." ), } assert _code_investigator_task_request(request) == { "repo": "pingcap/tidb", "commit_sha": sha, "focus_terms": ["partition state"], "mode": "targeted_follow_up", "parent_request_id": parent, "follow_up_reason": "User asked to inspect the state machine.", } request["prompt"] = request["prompt"].replace(parent, "code-invalid") assert _code_investigator_task_request(request) is None def test_code_investigator_follow_up_requires_a_fresh_current_turn_launch() -> None: """An earlier successful code run is lineage, never authorization.""" client = AgentEventBridge(workspace_id="ws-code-follow-up-gate") sha = "a" * 40 parent = "code-" + "1" * 32 task_input = { "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_FOLLOW_UP_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["Clear chain"],' f'"parent_request_id":"{parent}","follow_up_reason":"User requested the full chain."}}\n' "Inspect only the new source scope." ), } # A normal chat turn cannot turn an old manifest into a new child task. client._register_code_investigator_task( tool_id="ordinary-follow-up", title="Task", raw_input=task_input, session_id="session-1" ) ordinary = client._code_investigator_tasks["ordinary-follow-up"] assert ordinary["rejection_reason"] == "missing_or_invalid_ui_launch" assert ordinary["expected_commit_sha"] == "" # The same follow-up is allowed only after this turn's matching UI launch. client.authorize_code_diagnosis_launch( {"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-follow-up"} ) client._register_code_investigator_task( tool_id="authorized-follow-up", title="Task", raw_input=task_input, session_id="session-1" ) authorized = client._code_investigator_tasks["authorized-follow-up"] assert authorized["rejection_reason"] is None assert authorized["task_mode"] == "targeted_follow_up" assert authorized["expected_commit_sha"] == sha def test_code_investigator_launch_is_bound_to_its_agent_turn() -> None: from investibook.core.agent_runtime import ( build_code_diagnosis_launch, verify_code_diagnosis_launch, ) launch = build_code_diagnosis_launch( repo="pingcap/tidb", commit_sha="a" * 40, workspace_id="ws-code-run-binding", run_id="turn-original", ) assert verify_code_diagnosis_launch( launch + "\nInvestigate this.", workspace_id="ws-code-run-binding", expected_run_id="turn-original", ) is not None assert verify_code_diagnosis_launch( launch + "\nInvestigate this.", workspace_id="ws-code-run-binding", expected_run_id="turn-replayed", ) is None def test_code_investigator_permission_is_closed_without_current_launch(monkeypatch) -> None: """The permission callback is the pre-execution gate for an ask Task.""" client = AgentEventBridge(workspace_id="ws-code-permission") sha = "a" * 40 tool_call = types.SimpleNamespace( tool_call_id="permission-task", title="Task", raw_input={ "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["Clear"]}}\n' "Inspect source." ), }, ) monkeypatch.setattr(runtime_mod, "RequestPermissionResponse", lambda *, outcome: outcome) monkeypatch.setattr(runtime_mod, "DeniedOutcome", lambda *, outcome: ("denied", outcome)) monkeypatch.setattr(runtime_mod, "AllowedOutcome", lambda *, option_id, outcome: ("allowed", option_id, outcome)) option = types.SimpleNamespace(option_id="allow_once", kind="allow_once") denied = asyncio.run(client.request_permission([option], "session-1", tool_call)) assert denied == ("denied", "cancelled") client.authorize_code_diagnosis_launch( {"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-permission"} ) allowed = asyncio.run(client.request_permission([option], "session-1", tool_call)) assert allowed == ("allowed", "allow_once", "selected") # A launch is an atomic one-task authorization, not a reusable workspace state. second = asyncio.run(client.request_permission([option], "session-1", types.SimpleNamespace( tool_call_id="second-task", title=tool_call.title, raw_input=tool_call.raw_input, ))) assert second == ("denied", "cancelled") def test_code_investigator_permission_reuses_only_verified_sparse_task(monkeypatch) -> None: """OpenCode may omit Task.prompt in a later permission callback.""" client = AgentEventBridge(workspace_id="ws-code-sparse-permission") sha = "a" * 40 task_input = { "subagent_type": "code-investigator", "prompt": ( "INVESTIBOOK_CODE_CONTEXT_TASK_V1 " f'{{"repo":"pingcap/tidb","commit_sha":"{sha}","focus_terms":["slow log"]}}\n' "Inspect source." ), } client.authorize_code_diagnosis_launch( {"repo": "pingcap/tidb", "commit_sha": sha, "run_id": "turn-sparse"} ) client._register_code_investigator_task( tool_id="task-verified", title="Task", raw_input=task_input, session_id="session-1" ) monkeypatch.setattr(runtime_mod, "RequestPermissionResponse", lambda *, outcome: outcome) monkeypatch.setattr(runtime_mod, "DeniedOutcome", lambda *, outcome: ("denied", outcome)) monkeypatch.setattr(runtime_mod, "AllowedOutcome", lambda *, option_id, outcome: ("allowed", option_id, outcome)) option = types.SimpleNamespace(option_id="allow_once", kind="allow_once") sparse = types.SimpleNamespace( tool_call_id="task-verified", title="Task", raw_input={"subagent_type": "code-investigator"}, ) assert asyncio.run(client.request_permission([option], "session-1", sparse)) == ( "allowed", "allow_once", "selected" ) malformed = types.SimpleNamespace( tool_call_id="task-verified", title="Task", raw_input={"subagent_type": "code-investigator", "prompt": "not a machine header"}, ) assert asyncio.run(client.request_permission([option], "session-1", malformed)) == ( "denied", "cancelled" ) assert asyncio.run(client.request_permission([option], "other-session", sparse)) == ( "denied", "cancelled" ) def test_code_investigator_permission_defers_unregistered_sparse_task(monkeypatch) -> None: """The first ACP permission callback can arrive before Task.prompt.""" client = AgentEventBridge(workspace_id="ws-code-first-sparse-permission") monkeypatch.setattr(runtime_mod, "RequestPermissionResponse", lambda *, outcome: outcome) monkeypatch.setattr(runtime_mod, "DeniedOutcome", lambda *, outcome: ("denied", outcome)) monkeypatch.setattr(runtime_mod, "AllowedOutcome", lambda *, option_id, outcome: ("allowed", option_id, outcome)) option = types.SimpleNamespace(option_id="allow_once", kind="allow_once") sparse = types.SimpleNamespace( tool_call_id="task-first", title="Task", raw_input={"subagent_type": "code-investigator"}, ) assert asyncio.run(client.request_permission([option], "session-1", sparse)) == ( "allowed", "allow_once", "selected" ) assert client._code_investigator_tasks == {} def test_subagent_thought_meta_is_emitted_and_persisted(): client = AgentEventBridge(workspace_id="ws-subagent") client.start_assistant_message("assistant-subagent") meta = { "opencode/subagent": { "parentSessionId": "parent-session", "parentTaskCallId": "task-call", "subagentSessionId": "child-session", "subagentName": "evidence-agent", } } thought = _agent_thought_chunk( content={"text": "Checking TiDB CPU metrics"}, field_meta=meta, ) asyncio.run(client.session_update("session-1", thought)) event = client.queue.get_nowait() part = client._msg_parts[client._msg_parts_order[-1]] assert event["type"] == "thought" assert event["text"] == "Checking TiDB CPU metrics" assert event["meta"] == meta assert part["type"] == "reasoning" assert part["text"] == "Checking TiDB CPU metrics" assert part["meta"] == meta def test_subagent_message_meta_is_emitted_and_persisted_without_main_content(): client = AgentEventBridge(workspace_id="ws-subagent") client.start_assistant_message("assistant-subagent") meta = { "opencode/subagent": { "parentSessionId": "parent-session", "parentTaskCallId": "task-call", "subagentSessionId": "child-session", "subagentName": "evidence-agent", } } message = _agent_message_chunk( content={"text": "Evidence collection finished."}, field_meta=meta, ) asyncio.run(client.session_update("session-1", message)) event = client.queue.get_nowait() part = client._msg_parts[client._msg_parts_order[-1]] assert event["type"] == "text" assert event["text"] == "Evidence collection finished." assert event["meta"] == meta assert part["type"] == "text" assert part["text"] == "Evidence collection finished." assert part["meta"] == meta assert client._msg_text_parts == [] def test_failed_tool_progress_without_start_is_still_persistable(): client = AgentEventBridge(workspace_id="ws-refresh") client.start_assistant_message("assistant-refresh") progress = _tool_call_progress( tool_call_id="tool-1", title="Read file", raw_input={"path": "/tmp/missing"}, raw_output={"error": "file not found"}, status="completed", ) asyncio.run(client.session_update("session-1", progress)) part = client._msg_parts["tool-tool-1"] assert part["type"] == "tool" assert part["callID"] == "tool-1" assert part["state"]["status"] == "failed" assert part["state"]["output"] == {"error": "file not found"} def test_tool_progress_failed_status_is_terminal_failed(): client = AgentEventBridge(workspace_id="ws-refresh") client.start_assistant_message("assistant-refresh") start = _tool_call_start( tool_call_id="tool-2", title="Read file", raw_input={"path": "/tmp/missing"}, ) asyncio.run(client.session_update("session-1", start)) progress = _tool_call_progress( tool_call_id="tool-2", title="Read file", raw_input={"path": "/tmp/missing"}, raw_output={"message": "not found"}, status="failed", ) asyncio.run(client.session_update("session-1", progress)) part = client._msg_parts["tool-tool-2"] assert part["state"]["status"] == "failed" assert part["state"]["output"] == {"message": "not found"} def test_session_update_auto_writes_evidence_agent_after_collector_success(tmp_path, monkeypatch): monkeypatch.setattr(runtime_mod.cfg, "WORKSPACES_DIR", tmp_path) evidence_dir = tmp_path / "ws-collector" / "evidences" evidence_dir.mkdir(parents=True) summary_path = evidence_dir / "evidence_summary_for_agent.json" summary_path.write_text( json.dumps( { "metadata": { "cluster_id": "cluster-auto", "cluster": {"deployType": "dedicated"}, "time_range": {"start": "1", "end": "2"}, }, "critical_evidence_hints": [ {"id": "restart_signal_present", "summary": "restart evidence exists"} ], } ), encoding="utf-8", ) (evidence_dir / "workspace_evidence_report.json").write_text( json.dumps({"type": "workspace_evidence_report", "metadata": {}, "analysis": {}}), encoding="utf-8", ) (evidence_dir / "evidence_summary_for_agent.txt").write_text("summary", encoding="utf-8") (evidence_dir / "collection_errors.json").write_text( json.dumps({"errors": []}), encoding="utf-8", ) (evidence_dir / "metrics.json").write_text(json.dumps({"metrics": []}), encoding="utf-8") client = AgentEventBridge(workspace_id="ws-collector") client.start_assistant_message("assistant-collector") progress = _tool_call_progress( tool_call_id="tool-collector", title="bash", raw_input={"command": "python3 collect_evidence.py"}, raw_output={ "output": json.dumps( { "status": "ok", "run_dir": str(evidence_dir), "report_path": str(evidence_dir / "workspace_evidence_report.json"), "agent_summary_txt_path": str(evidence_dir / "evidence_summary_for_agent.txt"), "agent_summary_path": str(summary_path), "workspace_evidence_report_path": str(evidence_dir / "workspace_evidence_report.json"), "errors": [], } ), "metadata": {"exit": 0}, }, status="completed", ) asyncio.run(client.session_update("session-1", progress)) payload = json.loads((evidence_dir / "agent-runs" / "evidence-agent.json").read_text()) assert payload["agent"] == "evidence-agent" assert payload["scope"]["cluster_id"] == "cluster-auto" assert payload["scope"]["tier"] == "dedicated" assert payload["critical_evidence_hints"] == [ {"id": "restart_signal_present", "summary": "restart evidence exists"} ] def test_session_update_auto_writes_evidence_agent_with_descriptive_bash_title(tmp_path, monkeypatch): monkeypatch.setattr(runtime_mod.cfg, "WORKSPACES_DIR", tmp_path) evidence_dir = tmp_path / "ws-collector" / "evidences" evidence_dir.mkdir(parents=True) summary_path = evidence_dir / "evidence_summary_for_agent.json" summary_path.write_text( json.dumps( { "metadata": { "cluster_id": "cluster-auto", "cluster": {"deployType": "dedicated"}, "time_range": {"start": "1", "end": "2"}, } } ), encoding="utf-8", ) (evidence_dir / "workspace_evidence_report.json").write_text( json.dumps({"type": "workspace_evidence_report", "metadata": {}, "analysis": {}}), encoding="utf-8", ) (evidence_dir / "evidence_summary_for_agent.txt").write_text("summary", encoding="utf-8") (evidence_dir / "collection_errors.json").write_text( json.dumps({"errors": []}), encoding="utf-8", ) client = AgentEventBridge(workspace_id="ws-collector") client.start_assistant_message("assistant-collector") progress = _tool_call_progress( tool_call_id="tool-collector", title="Collect workspace evidence", raw_input={ "command": "python3 collect_evidence.py", "description": "Collect workspace evidence", }, raw_output={ "output": json.dumps( { "status": "ok", "run_dir": str(evidence_dir), "report_path": str(evidence_dir / "workspace_evidence_report.json"), "agent_summary_txt_path": str(evidence_dir / "evidence_summary_for_agent.txt"), "agent_summary_path": str(summary_path), "workspace_evidence_report_path": str(evidence_dir / "workspace_evidence_report.json"), "errors": [], } ), "metadata": {"exit": 0}, }, status="completed", ) asyncio.run(client.session_update("session-1", progress)) assert (evidence_dir / "agent-runs" / "evidence-agent.json").exists() def test_session_update_ignores_collector_output_for_other_workspace(tmp_path, monkeypatch): monkeypatch.setattr(runtime_mod.cfg, "WORKSPACES_DIR", tmp_path) evidence_dir = tmp_path / "ws-other" / "evidences" evidence_dir.mkdir(parents=True) summary_path = evidence_dir / "evidence_summary_for_agent.json" summary_path.write_text( json.dumps( { "metadata": { "cluster_id": "cluster-other", "cluster": {"deployType": "dedicated"}, "time_range": {"start": "1", "end": "2"}, } } ), encoding="utf-8", ) (evidence_dir / "workspace_evidence_report.json").write_text( json.dumps({"type": "workspace_evidence_report", "metadata": {}, "analysis": {}}), encoding="utf-8", ) (evidence_dir / "evidence_summary_for_agent.txt").write_text("summary", encoding="utf-8") (evidence_dir / "collection_errors.json").write_text( json.dumps({"errors": []}), encoding="utf-8", ) client = AgentEventBridge(workspace_id="ws-current") client.start_assistant_message("assistant-collector") progress = _tool_call_progress( tool_call_id="tool-collector", title="bash", raw_input={"command": "python3 collect_evidence.py"}, raw_output={ "output": json.dumps( { "status": "ok", "run_dir": str(evidence_dir), "report_path": str(evidence_dir / "workspace_evidence_report.json"), "agent_summary_txt_path": str(evidence_dir / "evidence_summary_for_agent.txt"), "agent_summary_path": str(summary_path), "workspace_evidence_report_path": str(evidence_dir / "workspace_evidence_report.json"), "errors": [], } ), "metadata": {"exit": 0}, }, status="completed", ) asyncio.run(client.session_update("session-1", progress)) assert not (evidence_dir / "agent-runs" / "evidence-agent.json").exists() def test_session_update_does_not_write_evidence_agent_when_root_artifacts_incomplete(tmp_path, monkeypatch): monkeypatch.setattr(runtime_mod.cfg, "WORKSPACES_DIR", tmp_path) evidence_dir = tmp_path / "ws-collector" / "evidences" evidence_dir.mkdir(parents=True) summary_path = evidence_dir / "evidence_summary_for_agent.json" summary_path.write_text( json.dumps( { "metadata": { "cluster_id": "cluster-auto", "cluster": {"deployType": "dedicated"}, "time_range": {"start": "1", "end": "2"}, } } ), encoding="utf-8", ) (evidence_dir / "workspace_evidence_report.json").write_text( json.dumps({"type": "workspace_evidence_report", "metadata": {}, "analysis": {}}), encoding="utf-8", ) client = AgentEventBridge(workspace_id="ws-collector") client.start_assistant_message("assistant-collector") progress = _tool_call_progress( tool_call_id="tool-collector", title="bash", raw_input={"command": "python3 collect_evidence.py"}, raw_output={ "output": json.dumps( { "status": "ok", "run_dir": str(evidence_dir), "report_path": str(evidence_dir / "workspace_evidence_report.json"), "agent_summary_txt_path": str(evidence_dir / "evidence_summary_for_agent.txt"), "agent_summary_path": str(summary_path), "workspace_evidence_report_path": str(evidence_dir / "workspace_evidence_report.json"), "errors": [], } ), "metadata": {"exit": 0}, }, status="completed", ) asyncio.run(client.session_update("session-1", progress)) assert not (evidence_dir / "agent-runs" / "evidence-agent.json").exists()