From 51dc49f0d8808c39d089861f61761bbb4db08179 Mon Sep 17 00:00:00 2001 From: Alex Date: Thu, 9 Jul 2026 00:11:27 +0100 Subject: [PATCH] feat: qol agent easier schema and better logs --- application/agents/base.py | 14 +-- application/agents/tool_executor.py | 45 +++++-- .../agents/tools/artifact_generator.py | 20 ++- application/agents/tools/code_executor.py | 34 ++++- application/agents/workflows/schemas.py | 5 + .../agents/workflows/workflow_engine.py | 64 +++++++++- .../storage/db/repositories/workflow_runs.py | 7 +- tests/agents/test_base_agent.py | 6 +- tests/agents/test_tool_executor.py | 44 ++++++- tests/agents/test_workflow_engine_coverage.py | 116 ++++++++++++++++++ .../tools/test_artifact_generator_unit.py | 42 +++++++ .../db/repositories/test_workflow_runs.py | 2 + tests/test_code_executor_tool.py | 42 +++++++ 13 files changed, 411 insertions(+), 30 deletions(-) diff --git a/application/agents/base.py b/application/agents/base.py index 0f6ea2b0..243eb3f6 100644 --- a/application/agents/base.py +++ b/application/agents/base.py @@ -4,7 +4,11 @@ import uuid from abc import ABC, abstractmethod from typing import Any, Dict, Generator, List, Optional -from application.agents.tool_executor import ToolExecutor +from application.agents.tool_executor import ( + ToolExecutor, + result_status, + truncate_tool_result, +) from application.core.json_schema_utils import ( JsonSchemaValidationError, normalize_json_schema_payload, @@ -298,12 +302,8 @@ class BaseAgent(ABC): "call_id": call_id, "action_name": pending.get("llm_name", pending["name"]), "arguments": args, - "result": ( - result_str[:50] + "..." - if len(result_str) > 50 - else result_str - ), - "status": "completed", + "result": truncate_tool_result(result_str), + "status": result_status(result), }, } diff --git a/application/agents/tool_executor.py b/application/agents/tool_executor.py index be392b30..34d6c58d 100644 --- a/application/agents/tool_executor.py +++ b/application/agents/tool_executor.py @@ -38,6 +38,32 @@ def _sanitize_tool_prefix(tool_name: Optional[str]) -> str: # program/secret is never written to logs even at DEBUG level. _LOG_VALUE_PREVIEW_LEN = 80 +# Longest tool result persisted on the message / streamed to the UI. The LLM +# and the ``tool_call_attempts`` journal always receive the full result; this +# only bounds the message JSONB copy. 50 chars hid every real error behind +# "...", making retry storms undiagnosable from the stored conversation. +PERSISTED_RESULT_MAX_LEN = 2000 + + +def truncate_tool_result(value: Any) -> Any: + """Bound a tool result for persistence/streaming; short values pass through unchanged.""" + text = value if isinstance(value, str) else str(value) + if len(text) <= PERSISTED_RESULT_MAX_LEN: + return value + return f"{text[:PERSISTED_RESULT_MAX_LEN]}..." + + +def result_status(result: Any) -> str: + """Derive the persisted status from a tool's result payload. + + Tools report failure in-band (``{"status": "error", ...}`` or an ``error`` + key) while the executor used to stamp every returned result ``completed``, + so the stored conversation showed failed calls as successes. + """ + if isinstance(result, dict) and (result.get("status") == "error" or result.get("error")): + return "error" + return "completed" + def _redact_args_for_log(args: Any) -> Any: """Truncate long string values so a code/body argument never lands in logs in full.""" @@ -605,6 +631,7 @@ class ToolExecutor: "action_name": llm_name, "arguments": call_args or {}, "result": f"Failed to parse tool call. Invalid tool name format: {llm_name}", + "status": "error", } # Journal the malformed call so it still shows up in tool analytics. if _record_proposed( @@ -639,6 +666,7 @@ class ToolExecutor: "action_name": llm_name, "arguments": call_args, "result": f"Tool with ID {tool_id} not found. Available tools: {list(tools_dict.keys())}", + "status": "error", } # Journal the unresolvable call so it still shows up in tool analytics. if _record_proposed( @@ -693,6 +721,7 @@ class ToolExecutor: error_message = f"Tool call arguments must be a JSON object, got {type(call_args).__name__}." tool_call_data["result"] = error_message tool_call_data["arguments"] = {} + tool_call_data["status"] = "error" if proposed_ok: _mark_failed(call_id, error_message, user_id=self.user) yield { @@ -749,9 +778,10 @@ class ToolExecutor: }, ) tool_call_data["result"] = error_message + tool_call_data["status"] = "error" if proposed_ok: _mark_failed(call_id, error_message, user_id=self.user) - yield {"type": "tool_call", "data": {**tool_call_data, "status": "error"}} + yield {"type": "tool_call", "data": {**tool_call_data}} self.tool_calls.append(tool_call_data) return error_message, call_id @@ -801,7 +831,10 @@ class ToolExecutor: result_full = str(result) tool_call_data["resolved_arguments"] = resolved_arguments tool_call_data["result_full"] = result_full - tool_call_data["result"] = f"{result_full[:50]}..." if len(result_full) > 50 else result_full + tool_call_data["result"] = truncate_tool_result(result_full) + # A tool that ran but reported failure in-band persists as ``error``, + # not ``completed`` -- the model saw an error and will likely retry. + tool_call_data["status"] = result_status(result) # Tool side effect has run; flip the journal row so the # message-finalize path can later confirm it. If the proposed @@ -824,7 +857,7 @@ class ToolExecutor: stream_tool_call_data = { key: value for key, value in tool_call_data.items() if key not in {"result_full", "resolved_arguments"} } - yield {"type": "tool_call", "data": {**stream_tool_call_data, "status": "completed"}} + yield {"type": "tool_call", "data": {**stream_tool_call_data}} self.tool_calls.append(tool_call_data) return result, call_id @@ -945,10 +978,8 @@ class ToolExecutor: "action_name": tool_call.get("action_name"), "arguments": tool_call.get("arguments"), "artifact_id": tool_call.get("artifact_id"), - "result": ( - f"{str(tool_call['result'])[:50]}..." if len(str(tool_call["result"])) > 50 else tool_call["result"] - ), - "status": "completed", + "result": truncate_tool_result(tool_call.get("result")), + "status": tool_call.get("status", "completed"), } for tool_call in self.tool_calls ] diff --git a/application/agents/tools/artifact_generator.py b/application/agents/tools/artifact_generator.py index a96425f7..8678a43c 100644 --- a/application/agents/tools/artifact_generator.py +++ b/application/agents/tools/artifact_generator.py @@ -217,6 +217,22 @@ _SCHEMAS: Dict[str, Dict[str, Any]] = { }, } +# Compact per-kind spec shapes surfaced in the tool metadata. Without this the +# model has only the validation errors to reverse-engineer the schema from and +# brute-forces spec shapes call after call. Keep in sync with ``_SCHEMAS`` +# (guarded by a test that checks every schema key appears here). +_SPEC_SYNOPSIS = ( + "Exact spec shape per kind (no other keys are accepted): " + 'presentation {"title"?, "slides": [{"title"?, "bullets"?: [str], "notes"?}]} · ' + 'document {"title"?, "sections": [{"heading"?, "paragraphs"?: [str]}]} · ' + 'spreadsheet {"sheets": [{"name"?, "rows": [[cell, ...]]}]} · ' + 'pdf {"title"?, "blocks": [{"type": "heading"|"paragraph", "text"}]} · ' + 'html {"title"?, "blocks": [...]} where each block is ' + '{"type": "heading", "text", "level"?: 1-3} | {"type": "paragraph", "text"} | ' + '{"type": "list", "items": [str], "ordered"?: bool} | ' + '{"type": "table", "rows": [[str]], "headers"?: [str]} | {"type": "code", "text"}' +) + # FIXED renderer programs. Each reads ``spec.json`` from the workspace as DATA # and writes ``out.``. The spec is NEVER string-interpolated into the # program; ``{spec_path}``/``{out_path}`` are server-controlled path literals. @@ -426,7 +442,7 @@ class ArtifactGeneratorTool(Tool): ), }, "title": {"type": "string", "description": "Optional artifact title."}, - "spec": {"type": "object", "description": "Document spec matching the kind's schema."}, + "spec": {"type": "object", "description": _SPEC_SYNOPSIS}, }, "required": ["kind", "spec"], }, @@ -466,7 +482,7 @@ class ArtifactGeneratorTool(Tool): "description": "Artifact to rewrite; accepts the short ref like `A1` " "returned by a previous artifact action, or the full artifact id.", }, - "spec": {"type": "object", "description": "Replacement spec matching the kind's schema."}, + "spec": {"type": "object", "description": f"Replacement spec. {_SPEC_SYNOPSIS}"}, }, "required": ["id", "spec"], }, diff --git a/application/agents/tools/code_executor.py b/application/agents/tools/code_executor.py index 260128b9..0bcc7910 100644 --- a/application/agents/tools/code_executor.py +++ b/application/agents/tools/code_executor.py @@ -77,6 +77,33 @@ class CodeExecutorTool(Tool): # ------------------------------------------------------------------ # Tool ABC # ------------------------------------------------------------------ + @staticmethod + def _environment_note() -> str: + """Backend-specific note on what the sandbox has preinstalled. + + Without this the model discovers the environment by failing: importing + pandas on a bare image, or pip-installing libraries that are already + baked in. Keep the package lists in sync with deployment/sandbox/Dockerfile + (jupyter) and scripts/build_daytona_snapshot.py (daytona snapshot). + """ + backend = str(getattr(settings, "SANDBOX_BACKEND", "jupyter") or "jupyter").lower() + if backend == "daytona": + if getattr(settings, "DAYTONA_SNAPSHOT", None): + return ( + "Preinstalled beyond the stdlib: python-pptx, python-docx, openpyxl, " + "reportlab, lxml, pillow. pip install anything else from within the code " + "before importing it." + ) + return ( + "Only the Python stdlib is preinstalled. pip install any third-party " + "package (pandas, python-docx, ...) from within the code before importing it." + ) + return ( + "Preinstalled beyond the stdlib: pandas, matplotlib, python-pptx, python-docx, " + "openpyxl, reportlab. pip install anything else from within the code before " + "importing it." + ) + def get_actions_metadata(self) -> List[Dict[str, Any]]: """Return JSON metadata describing the ``run_code`` action for tool schemas.""" return [ @@ -88,7 +115,8 @@ class CodeExecutorTool(Tool): "files under `tmp/`, or pass `outputs` to save only specific files); only a compact " "summary (output tail + artifact references) is returned, never raw bytes. " "Each call is capped at ~60s of wall-clock; for longer work, start it in the " - "background and poll with additional run_code calls (use persist=true to keep state)." + "background and poll with additional run_code calls (use persist=true to keep state). " + + self._environment_note() ), "active": True, "require_approval": self._require_approval, @@ -105,7 +133,9 @@ class CodeExecutorTool(Tool): "items": {"type": "string"}, "description": "Files to materialize into the workspace; each accepts the short " "ref like `A1` returned by a previous artifact action, a full artifact id, or " - "the name/id of a file the user attached to this conversation.", + "the name/id of a file the user attached to this conversation. Each is staged " + "at `inputs/` before the code runs — read it from that path (the " + "result's `inputs_loaded` echoes the exact staged paths).", }, "outputs": { "type": "array", diff --git a/application/agents/workflows/schemas.py b/application/agents/workflows/schemas.py index f5a806c4..73d2f9b1 100644 --- a/application/agents/workflows/schemas.py +++ b/application/agents/workflows/schemas.py @@ -165,8 +165,13 @@ class NodeExecutionLog(BaseModel): status: ExecutionStatus started_at: datetime completed_at: Optional[datetime] = None + duration_ms: Optional[int] = None error: Optional[str] = None + # The node's state DELTA (keys it added or changed), not the full state: + # point-in-time state is the merge of deltas up to this step. state_snapshot: Dict[str, Any] = Field(default_factory=dict) + # Compact per-node tool-call summary: [{tool_name, action_name, status}]. + tool_calls: List[Dict[str, Any]] = Field(default_factory=list) class WorkflowRunCreate(BaseModel): diff --git a/application/agents/workflows/workflow_engine.py b/application/agents/workflows/workflow_engine.py index adc20b83..04810fb9 100644 --- a/application/agents/workflows/workflow_engine.py +++ b/application/agents/workflows/workflow_engine.py @@ -69,6 +69,9 @@ class WorkflowEngine: # artifact's parent; mint one when the caller has not supplied a # pre-created ``workflow_runs`` id so the engine is self-contained. self.workflow_run_id: str = workflow_run_id or str(uuid.uuid4()) + # Per-node tool-call summary collected by the node executors and folded + # into the step log; reset before each node runs. + self._last_node_tool_calls: List[Dict[str, Any]] = [] # False when no ``workflow_runs`` row backs ``workflow_run_id`` (unsaved / # embedded draft), so run-scoped artifacts must NOT be persisted as orphans. # Defaults True; WorkflowAgent flips it off on the draft path. @@ -115,6 +118,12 @@ class WorkflowEngine: return current_node_id: Optional[str] = start_node.id steps = 0 + # Snapshots are stored as per-node DELTAS: full state copied into + # every step grows the run row O(n^2) and repeats every upstream + # output verbatim. Point-in-time state = merge of deltas up to a step. + # The empty baseline attributes the pre-loop initialization (query, + # input documents) to the first step so steps alone reconstruct state. + pre_state: Dict[str, Any] = {} while current_node_id and steps < self.MAX_EXECUTION_STEPS: node = self.graph.get_node_by_id(current_node_id) @@ -122,6 +131,7 @@ class WorkflowEngine: yield {"type": "error", "error": f"Node {current_node_id} not found."} break log_entry = self._create_log_entry(node) + self._last_node_tool_calls = [] yield { "type": "workflow_step", @@ -134,7 +144,7 @@ class WorkflowEngine: try: yield from self._execute_node(node) log_entry["status"] = ExecutionStatus.COMPLETED.value - log_entry["completed_at"] = datetime.now(timezone.utc) + self._finalize_log_entry(log_entry, pre_state) output_key = f"node_{node.id}_output" node_output = self.state.get(output_key) @@ -145,15 +155,14 @@ class WorkflowEngine: "node_type": node.type.value, "node_title": node.title, "status": "completed", - "state_snapshot": dict(self.state), + "state_snapshot": log_entry["state_snapshot"], "output": node_output, } except Exception as e: logger.error(f"Error executing node {node.id}: {e}", exc_info=True) log_entry["status"] = ExecutionStatus.FAILED.value log_entry["error"] = str(e) - log_entry["completed_at"] = datetime.now(timezone.utc) - log_entry["state_snapshot"] = dict(self.state) + self._finalize_log_entry(log_entry, pre_state) self.execution_log.append(log_entry) user_friendly_error = sanitize_api_error(e) @@ -163,13 +172,13 @@ class WorkflowEngine: "node_type": node.type.value, "node_title": node.title, "status": "failed", - "state_snapshot": dict(self.state), + "state_snapshot": log_entry["state_snapshot"], "error": user_friendly_error, } yield {"type": "error", "error": user_friendly_error} break - log_entry["state_snapshot"] = dict(self.state) self.execution_log.append(log_entry) + pre_state = dict(self.state) if node.type == NodeType.END: break @@ -200,6 +209,37 @@ class WorkflowEngine: "state_snapshot": {}, } + def _finalize_log_entry(self, log_entry: Dict[str, Any], pre_state: Dict[str, Any]) -> None: + """Stamp completion time, duration, the node's state delta, and its tool-call summary.""" + completed_at = datetime.now(timezone.utc) + log_entry["completed_at"] = completed_at + log_entry["duration_ms"] = int((completed_at - log_entry["started_at"]).total_seconds() * 1000) + log_entry["state_snapshot"] = self._state_delta(pre_state) + if self._last_node_tool_calls: + log_entry["tool_calls"] = list(self._last_node_tool_calls) + + def _state_delta(self, previous: Dict[str, Any]) -> Dict[str, Any]: + """Keys this node added or changed. Deleted keys are not tracked; nothing deletes state today.""" + return { + key: value + for key, value in self.state.items() + if key not in previous or previous[key] != value + } + + @staticmethod + def _summarize_tool_calls(node_agent: Any) -> List[Dict[str, Any]]: + """Compact per-node tool-call summary for the run record (no arguments/results).""" + executor = getattr(node_agent, "tool_executor", None) + calls = getattr(executor, "tool_calls", None) or [] + return [ + { + "tool_name": call.get("tool_name"), + "action_name": call.get("action_name"), + "status": call.get("status", "completed"), + } + for call in calls + ] + def _get_next_node_id(self, current_node_id: str) -> Optional[str]: node = self.graph.get_node_by_id(current_node_id) edges = self.graph.get_outgoing_edges(current_node_id) @@ -370,6 +410,7 @@ class WorkflowEngine: if node_config.stream_to_user: self._has_streamed = True + self._last_node_tool_calls = self._summarize_tool_calls(node_agent) full_response = "".join(full_response_parts).strip() output_value: Any = full_response if has_structured_response: @@ -436,6 +477,15 @@ class WorkflowEngine: manager.put_file(session_id, "state.json", state_json) pre_signatures = snapshot_signatures(manager, session_id) result = manager.exec(session_id, code, timeout=timeout) + # Code nodes execute the sandbox directly (no tool_executor); record the + # run as a synthetic tool call so the step log shows it like agent nodes. + self._last_node_tool_calls = [ + { + "tool_name": "code_executor", + "action_name": "run_code", + "status": "completed" if result.ok else "error", + } + ] if self.run_persisted: artifacts = capture_artifacts( manager, @@ -1083,8 +1133,10 @@ class WorkflowEngine: status=ExecutionStatus(log["status"]), started_at=log["started_at"], completed_at=log.get("completed_at"), + duration_ms=log.get("duration_ms"), error=log.get("error"), state_snapshot=log.get("state_snapshot", {}), + tool_calls=log.get("tool_calls", []), ) for log in self.execution_log ] diff --git a/application/storage/db/repositories/workflow_runs.py b/application/storage/db/repositories/workflow_runs.py index 25e105e9..a3ee7815 100644 --- a/application/storage/db/repositories/workflow_runs.py +++ b/application/storage/db/repositories/workflow_runs.py @@ -103,7 +103,12 @@ class WorkflowRunsRepository: .values( status="failed", ended_at=func.now(), - result={"error": "Run did not complete (timed out or the client disconnected)."}, + # failed_reason marks the reap so readers know ended_at is the + # reap time, not when the run actually died (duration is bogus). + result={ + "error": "Run did not complete (timed out or the client disconnected).", + "failed_reason": "stale_reaper", + }, ) ) res = self._conn.execute(stmt) diff --git a/tests/agents/test_base_agent.py b/tests/agents/test_base_agent.py index 7f8a8b34..947affec 100644 --- a/tests/agents/test_base_agent.py +++ b/tests/agents/test_base_agent.py @@ -559,20 +559,22 @@ class TestBaseAgentToolExecution: ): agent = ClassicAgent(**agent_base_params) + from application.agents.tool_executor import PERSISTED_RESULT_MAX_LEN + agent.tool_calls = [ { "tool_name": "test_tool", "call_id": "123", "action_name": "action", "arguments": {}, - "result": "a" * 100, + "result": "a" * (PERSISTED_RESULT_MAX_LEN + 100), } ] truncated = agent._get_truncated_tool_calls() assert len(truncated) == 1 - assert len(truncated[0]["result"]) <= 53 + assert len(truncated[0]["result"]) == PERSISTED_RESULT_MAX_LEN + 3 assert truncated[0]["result"].endswith("...") diff --git a/tests/agents/test_tool_executor.py b/tests/agents/test_tool_executor.py index 6996570a..237c7562 100644 --- a/tests/agents/test_tool_executor.py +++ b/tests/agents/test_tool_executor.py @@ -965,6 +965,8 @@ class TestToolExecutorExecute: assert "completed" in statuses def test_get_truncated_tool_calls(self): + from application.agents.tool_executor import PERSISTED_RESULT_MAX_LEN + executor = ToolExecutor() executor.tool_calls = [ { @@ -973,13 +975,49 @@ class TestToolExecutorExecute: "action_name": "act", "arguments": {}, "result": "A" * 100, + }, + { + "tool_name": "test", + "call_id": "2", + "action_name": "act", + "arguments": {}, + "result": "B" * (PERSISTED_RESULT_MAX_LEN + 100), + }, + ] + + truncated = executor.get_truncated_tool_calls() + assert len(truncated) == 2 + # Short results persist verbatim so stored errors stay diagnosable. + assert truncated[0]["result"] == "A" * 100 + assert truncated[0]["status"] == "completed" + # Oversize results are bounded for the message JSONB copy. + assert len(truncated[1]["result"]) == PERSISTED_RESULT_MAX_LEN + 3 + assert truncated[1]["result"].endswith("...") + + def test_get_truncated_tool_calls_preserves_error_status(self): + executor = ToolExecutor() + executor.tool_calls = [ + { + "tool_name": "test", + "call_id": "1", + "action_name": "act", + "arguments": {}, + "result": "boom", + "status": "error", } ] truncated = executor.get_truncated_tool_calls() - assert len(truncated) == 1 - assert len(truncated[0]["result"]) <= 53 - assert truncated[0]["status"] == "completed" + assert truncated[0]["status"] == "error" + + def test_result_status_reflects_in_band_tool_errors(self): + from application.agents.tool_executor import result_status + + assert result_status({"status": "error", "error": "invalid spec"}) == "error" + assert result_status({"error": "input artifact A9 not found"}) == "error" + assert result_status({"status": "ok", "artifact_id": "x"}) == "completed" + assert result_status({"error": None, "status": "ok"}) == "completed" + assert result_status("plain text result") == "completed" def test_tool_caching(self, mock_tool_manager, monkeypatch): executor = ToolExecutor(user="test_user") diff --git a/tests/agents/test_workflow_engine_coverage.py b/tests/agents/test_workflow_engine_coverage.py index 709a4050..0aed618b 100644 --- a/tests/agents/test_workflow_engine_coverage.py +++ b/tests/agents/test_workflow_engine_coverage.py @@ -846,3 +846,119 @@ class TestWorkflowBranchEndsNonEndNode: events = list(engine.execute({}, "q")) # Should complete without crash, branch ended warning logged assert len(events) > 0 + + +class TestStepTelemetry: + """Steps persist state DELTAS plus duration_ms and a per-node tool summary.""" + + def _run_state_workflow(self): + nodes = [ + _make_node("n1", NodeType.START), + _make_node( + "n2", + NodeType.STATE, + "Set", + config={ + "config": { + "operations": [ + {"expression": "1 + 1", "target_variable": "x"} + ] + } + }, + ), + _make_node("n3", NodeType.END, "End", config={"config": {}}), + ] + edges = [_make_edge("e1", "n1", "n2"), _make_edge("e2", "n2", "n3")] + engine = WorkflowEngine(_make_graph(nodes, edges), _make_agent()) + events = list(engine.execute({}, "q")) + return engine, events + + @pytest.mark.unit + def test_step_snapshots_are_deltas_not_full_state(self): + engine, events = self._run_state_workflow() + by_node = {log["node_id"]: log for log in engine.execution_log} + + # The start node's delta carries the initial state it set... + assert "query" in by_node["n1"]["state_snapshot"] + # ...and later nodes repeat none of it, only what they changed. + assert by_node["n2"]["state_snapshot"] == {"x": 2} + assert by_node["n3"]["state_snapshot"] == {} + + # The streamed step events carry the same deltas. + streamed = { + e["node_id"]: e + for e in events + if e.get("type") == "workflow_step" and e.get("status") == "completed" + } + assert streamed["n2"]["state_snapshot"] == {"x": 2} + assert "query" not in streamed["n2"]["state_snapshot"] + + @pytest.mark.unit + def test_steps_carry_duration_ms(self): + engine, _ = self._run_state_workflow() + for log in engine.execution_log: + assert isinstance(log["duration_ms"], int) + assert log["duration_ms"] >= 0 + + summary = engine.get_execution_summary() + assert all(step.duration_ms is not None for step in summary) + + @pytest.mark.unit + def test_failed_step_snapshot_is_a_delta_too(self): + nodes = [ + _make_node("n1", NodeType.START), + _make_node( + "n2", + NodeType.STATE, + "Bad", + config={ + "config": { + "operations": [ + {"expression": "bad!!!", "target_variable": "x"} + ] + } + }, + ), + ] + edges = [_make_edge("e1", "n1", "n2")] + engine = WorkflowEngine(_make_graph(nodes, edges), _make_agent()) + events = list(engine.execute({}, "q")) + + failed = [e for e in events if e.get("status") == "failed"] + assert failed and "query" not in failed[0]["state_snapshot"] + failed_log = engine.execution_log[-1] + assert failed_log["status"] == "failed" + assert failed_log["duration_ms"] >= 0 + + @pytest.mark.unit + def test_summarize_tool_calls_compacts_executor_calls(self): + engine = WorkflowEngine(_make_graph([], []), _make_agent()) + node_agent = MagicMock() + node_agent.tool_executor.tool_calls = [ + { + "tool_name": "artifact_generator", + "action_name": "create_artifact", + "arguments": {"kind": "html"}, + "result": "x" * 5000, + "status": "error", + }, + { + "tool_name": "artifact_generator", + "action_name": "create_artifact", + "arguments": {"kind": "html"}, + "result": "ok", + }, + ] + summary = engine._summarize_tool_calls(node_agent) + assert summary == [ + { + "tool_name": "artifact_generator", + "action_name": "create_artifact", + "status": "error", + }, + { + "tool_name": "artifact_generator", + "action_name": "create_artifact", + "status": "completed", + }, + ] diff --git a/tests/agents/tools/test_artifact_generator_unit.py b/tests/agents/tools/test_artifact_generator_unit.py index 1b04e35f..cdd73938 100644 --- a/tests/agents/tools/test_artifact_generator_unit.py +++ b/tests/agents/tools/test_artifact_generator_unit.py @@ -363,3 +363,45 @@ def test_html_renderer_escapes_spec_markup(): assert "" in html_doc assert "

" in html_doc and "

" in html_doc assert "" in html_doc and "
" in html_doc
+
+
+# ---------------------------------------------------------------------------
+# Tool metadata: spec shapes are surfaced to the model
+# ---------------------------------------------------------------------------
+
+
+def _collect_schema_keys(schema, keys):
+    if isinstance(schema, dict):
+        for prop in schema.get("properties", {}):
+            keys.add(prop)
+        for sub in schema.values():
+            _collect_schema_keys(sub, keys)
+    elif isinstance(schema, list):
+        for sub in schema:
+            _collect_schema_keys(sub, keys)
+
+
+def test_spec_synopsis_covers_every_schema_key():
+    """Every kind and property in _SCHEMAS must appear in the metadata synopsis.
+
+    The synopsis is a hand-written mirror of _SCHEMAS; this guards drift when a
+    kind or key is added without updating what the model is told.
+    """
+    from application.agents.tools.artifact_generator import _SCHEMAS, _SPEC_SYNOPSIS
+
+    for kind, schema in _SCHEMAS.items():
+        assert kind in _SPEC_SYNOPSIS
+        keys: set = set()
+        _collect_schema_keys(schema, keys)
+        for key in keys:
+            assert key in _SPEC_SYNOPSIS, f"schema key {key!r} of kind {kind!r} missing from synopsis"
+
+
+def test_create_and_rewrite_metadata_embed_spec_synopsis():
+    from application.agents.tools.artifact_generator import _SPEC_SYNOPSIS
+
+    actions = {a["name"]: a for a in _tool().get_actions_metadata()}
+    create_spec = actions["create_artifact"]["parameters"]["properties"]["spec"]
+    rewrite_spec = actions["rewrite_artifact"]["parameters"]["properties"]["spec"]
+    assert _SPEC_SYNOPSIS in create_spec["description"]
+    assert _SPEC_SYNOPSIS in rewrite_spec["description"]
diff --git a/tests/storage/db/repositories/test_workflow_runs.py b/tests/storage/db/repositories/test_workflow_runs.py
index 84f5c7e1..722d25d4 100644
--- a/tests/storage/db/repositories/test_workflow_runs.py
+++ b/tests/storage/db/repositories/test_workflow_runs.py
@@ -115,6 +115,8 @@ class TestMarkStaleRunningFailed:
         assert fetched["status"] == "failed"
         assert fetched["ended_at"] is not None
         assert "did not complete" in (fetched["result"] or {}).get("error", "")
+        # ended_at is the reap time, not when the run died; the marker says so.
+        assert (fetched["result"] or {}).get("failed_reason") == "stale_reaper"
 
     def test_leaves_recent_running_and_terminal_runs(self, pg_conn):
         wf = _wf(pg_conn)
diff --git a/tests/test_code_executor_tool.py b/tests/test_code_executor_tool.py
index f2a33ecf..496af9f1 100644
--- a/tests/test_code_executor_tool.py
+++ b/tests/test_code_executor_tool.py
@@ -479,3 +479,45 @@ def test_materialize_inputs_dedupes_same_filename(monkeypatch):
     assert set(manager.put_files) == {"inputs/seed.csv", "inputs/seed-2.csv"}
     # Each path holds its own artifact's bytes (no clobber).
     assert manager.put_files["inputs/seed.csv"] != manager.put_files["inputs/seed-2.csv"]
+
+
+# ---------------------------------------------------------------------------
+# Metadata: workspace contract and preinstalled packages
+# ---------------------------------------------------------------------------
+
+
+def test_inputs_metadata_names_the_staging_path():
+    props = _tool().get_actions_metadata()[0]["parameters"]["properties"]
+    assert "inputs/" in props["inputs"]["description"]
+    assert "inputs_loaded" in props["inputs"]["description"]
+
+
+def test_description_lists_jupyter_preinstalled_packages(monkeypatch):
+    from application.core import settings as settings_module
+
+    monkeypatch.setattr(settings_module.settings, "SANDBOX_BACKEND", "jupyter", raising=False)
+    desc = _tool().get_actions_metadata()[0]["description"]
+    for pkg in ("pandas", "matplotlib", "python-docx"):
+        assert pkg in desc
+
+
+def test_description_warns_bare_daytona_image(monkeypatch):
+    from application.core import settings as settings_module
+
+    monkeypatch.setattr(settings_module.settings, "SANDBOX_BACKEND", "daytona", raising=False)
+    monkeypatch.setattr(settings_module.settings, "DAYTONA_SNAPSHOT", None, raising=False)
+    desc = _tool().get_actions_metadata()[0]["description"]
+    assert "Only the Python stdlib is preinstalled" in desc
+
+
+def test_description_lists_daytona_snapshot_packages(monkeypatch):
+    from application.core import settings as settings_module
+
+    monkeypatch.setattr(settings_module.settings, "SANDBOX_BACKEND", "daytona", raising=False)
+    monkeypatch.setattr(
+        settings_module.settings, "DAYTONA_SNAPSHOT", "docsgpt-artifacts-py312", raising=False
+    )
+    desc = _tool().get_actions_metadata()[0]["description"]
+    assert "python-pptx" in desc
+    # The snapshot bakes render libs only, not pandas.
+    assert "pandas" not in desc