feat: qol agent easier schema and better logs

This commit is contained in:
Alex committed 2026-07-09 00:11:27 +01:00
1 parent 7f57e06531
commit 51dc49f0d8
13 files changed
+411 -30

No files matched your search

+7 -7
View File
@@ -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),
},
}
+38 -7
View File
@@ -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
]
+18 -2
View File
@@ -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.<ext>``. 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"],
},
+32 -2
View File
@@ -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/<filename>` before the code runs — read it from that path (the "
"result's `inputs_loaded` echoes the exact staged paths).",
},
"outputs": {
"type": "array",
+5
View File
@@ -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):
@@ -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
]
@@ -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)
+4 -2
View File
@@ -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("...")
+41 -3
View File
@@ -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")
@@ -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",
},
]
@@ -363,3 +363,45 @@ def test_html_renderer_escapes_spec_markup():
assert "<!doctype html>" in html_doc
assert "<p>" in html_doc and "</p>" in html_doc
assert "<table>" in html_doc and "<pre><code>" 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"]
@@ -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)
+42
View File
@@ -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/<filename>" 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