diff --git a/application/agents/workflows/schemas.py b/application/agents/workflows/schemas.py index 73d2f9b1..c8b2651f 100644 --- a/application/agents/workflows/schemas.py +++ b/application/agents/workflows/schemas.py @@ -168,8 +168,9 @@ class NodeExecutionLog(BaseModel): 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) + # point-in-time state is the merge of deltas up to this step. Runs + # persisted before the rename carry this as ``state_snapshot``. + state_delta: 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) diff --git a/application/agents/workflows/workflow_engine.py b/application/agents/workflows/workflow_engine.py index 04810fb9..cf426d54 100644 --- a/application/agents/workflows/workflow_engine.py +++ b/application/agents/workflows/workflow_engine.py @@ -155,7 +155,7 @@ class WorkflowEngine: "node_type": node.type.value, "node_title": node.title, "status": "completed", - "state_snapshot": log_entry["state_snapshot"], + "state_delta": log_entry["state_delta"], "output": node_output, } except Exception as e: @@ -172,7 +172,7 @@ class WorkflowEngine: "node_type": node.type.value, "node_title": node.title, "status": "failed", - "state_snapshot": log_entry["state_snapshot"], + "state_delta": log_entry["state_delta"], "error": user_friendly_error, } yield {"type": "error", "error": user_friendly_error} @@ -206,7 +206,7 @@ class WorkflowEngine: "completed_at": None, "status": ExecutionStatus.RUNNING.value, "error": None, - "state_snapshot": {}, + "state_delta": {}, } def _finalize_log_entry(self, log_entry: Dict[str, Any], pre_state: Dict[str, Any]) -> None: @@ -214,7 +214,7 @@ class WorkflowEngine: 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) + log_entry["state_delta"] = self._state_delta(pre_state) if self._last_node_tool_calls: log_entry["tool_calls"] = list(self._last_node_tool_calls) @@ -334,6 +334,11 @@ class WorkflowEngine: f'Model "{node_model_id}" does not support structured output for node "{node.title}"' ) + node_prompt = node_config.system_prompt + doc_manifest = self._node_document_manifest(node_config) + if doc_manifest: + node_prompt = f"{node_prompt}\n\n{doc_manifest}" if node_prompt else doc_manifest + factory_kwargs = { "agent_type": node_config.agent_type, "endpoint": self.agent.endpoint, @@ -342,7 +347,7 @@ class WorkflowEngine: "model_user_id": getattr(self.agent, "model_user_id", None), "api_key": node_api_key, "tool_ids": node_config.tools, - "prompt": node_config.system_prompt, + "prompt": node_prompt, "chat_history": self.agent.chat_history, "decoded_token": self.agent.decoded_token, "json_schema": node_json_schema, @@ -610,6 +615,54 @@ class WorkflowEngine: loaded.append(rel_path) return loaded + def _node_document_manifest(self, node_config: AgentNodeConfig) -> str: + """One-line-per-document manifest of the node's selected input documents. + + Appended to the node's system prompt so the model knows the concrete + refs (``A1``) and filenames it can pass to document/code tools — + without it, models guess placeholder names ("attached_file") or paste + file contents inline. Scoped to THIS node's ``input_documents`` + selection, so it never widens per-node document access. Best-effort: + a resolution failure drops the manifest, never the node. + """ + from application.agents.tools.artifact_ref import make_ref, resolve_artifact_id + from application.storage.db.repositories.artifacts import ArtifactsRepository + from application.storage.db.session import db_readonly + + try: + raw_ids = self._resolve_input_artifact_ids(node_config.input_documents) + if not raw_ids: + return "" + lines: List[str] = [] + with db_readonly() as conn: + repo = ArtifactsRepository(conn) + for raw in raw_ids: + artifact_id = resolve_artifact_id(repo, raw, workflow_run_id=self.workflow_run_id) + artifact = ( + repo.get_artifact_in_parent(artifact_id, workflow_run_id=self.workflow_run_id) + if artifact_id is not None + else None + ) + if artifact is None: + continue + version = repo.get_version(artifact_id, artifact["current_version"]) or {} + ref_seq = (artifact.get("metadata") or {}).get("ref_seq") + # Legacy artifacts predate ref_seq; the full id works in every tool. + handle = make_ref(int(ref_seq)) if ref_seq else str(artifact_id) + filename = version.get("filename") or artifact.get("title") or str(artifact_id) + mime = version.get("mime_type") or "application/octet-stream" + lines.append(f"- {handle}: {filename} ({mime})") + except Exception: + logger.exception("Node document manifest failed; continuing without it") + return "" + if not lines: + return "" + return ( + "## Input documents for this node\n" + "These files are staged for this node. Pass a ref (e.g. A1) or the " + "filename to a document or code tool to read the file:\n" + "\n".join(lines) + ) + def _materialize_node_attachments( self, node_config: AgentNodeConfig, @@ -1135,7 +1188,7 @@ class WorkflowEngine: completed_at=log.get("completed_at"), duration_ms=log.get("duration_ms"), error=log.get("error"), - state_snapshot=log.get("state_snapshot", {}), + state_delta=log.get("state_delta", {}), tool_calls=log.get("tool_calls", []), ) for log in self.execution_log diff --git a/frontend/src/agents/workflow/WorkflowBuilder.tsx b/frontend/src/agents/workflow/WorkflowBuilder.tsx index 83bd5c8a..8693d62e 100644 --- a/frontend/src/agents/workflow/WorkflowBuilder.tsx +++ b/frontend/src/agents/workflow/WorkflowBuilder.tsx @@ -3030,6 +3030,7 @@ function WorkflowBuilderInner() { n.id === step.nodeId); const displayName = node?.title || node?.data?.title || step.nodeTitle; - const stateVars = step.stateSnapshot - ? Object.entries(step.stateSnapshot).filter( + const stateVars = step.stateDelta + ? Object.entries(step.stateDelta).filter( ([key]) => !['query', 'chat_history'].includes(key), ) : []; @@ -253,10 +253,12 @@ function RunArtifactsSection({ workflowRunId, isOpen, onToggle, + runInProgress = false, }: { workflowRunId: string; isOpen: boolean; onToggle: () => void; + runInProgress?: boolean; }) { return (
@@ -289,7 +291,12 @@ function RunArtifactsSection({ >
- {isOpen && } + {isOpen && ( + + )}
@@ -687,9 +694,10 @@ export default function WorkflowPreview({ /> )} - {/* Run artifacts (only once a persisted run id is known - and the run is no longer streaming) */} - {query.workflowRunId && !isStreamingLastQuery && ( + {/* Run artifacts — shown as soon as a persisted run id + is known; mid-run it lists artifacts as nodes produce + them and refetches when the run completes. */} + {query.workflowRunId && ( )} diff --git a/frontend/src/agents/workflow/WorkflowRunArtifacts.tsx b/frontend/src/agents/workflow/WorkflowRunArtifacts.tsx index 53703468..f395f9c2 100644 --- a/frontend/src/agents/workflow/WorkflowRunArtifacts.tsx +++ b/frontend/src/agents/workflow/WorkflowRunArtifacts.tsx @@ -21,11 +21,15 @@ interface RunArtifactSummary { interface WorkflowRunArtifactsProps { workflowRunId: string; + /** True while the run is still streaming; flipping to false refetches so + * a panel opened mid-run picks up the final artifact list. */ + inProgress?: boolean; } /** List a workflow run's produced artifacts, with click-through preview + download. */ export default function WorkflowRunArtifacts({ workflowRunId, + inProgress = false, }: WorkflowRunArtifactsProps) { const token = useSelector(selectToken); const [artifacts, setArtifacts] = useState(null); @@ -77,7 +81,7 @@ export default function WorkflowRunArtifacts({ useEffect(() => { const cleanup = loadList(); return cleanup; - }, [loadList]); + }, [loadList, inProgress]); const fetchDetail = useCallback( (artifactId: string) => { @@ -147,7 +151,9 @@ export default function WorkflowRunArtifacts({ if (!artifacts || artifacts.length === 0) { return (
- No artifacts produced by this run. + {inProgress + ? 'Run in progress — artifacts appear here as nodes produce them.' + : 'No artifacts produced by this run.'}
); } diff --git a/frontend/src/agents/workflow/workflowPreviewSlice.ts b/frontend/src/agents/workflow/workflowPreviewSlice.ts index b7888885..b2aa88d9 100644 --- a/frontend/src/agents/workflow/workflowPreviewSlice.ts +++ b/frontend/src/agents/workflow/workflowPreviewSlice.ts @@ -13,7 +13,7 @@ export interface WorkflowExecutionStep { reasoning?: string; startedAt?: number; completedAt?: number; - stateSnapshot?: Record; + stateDelta?: Record; output?: unknown; error?: string; } @@ -212,7 +212,8 @@ export const fetchWorkflowPreviewAnswer = createAsyncThunk< nodeTitle: data.node_title, status: data.status, reasoning: data.reasoning, - stateSnapshot: data.state_snapshot, + // Older streams (pre-rename) still say state_snapshot. + stateDelta: data.state_delta ?? data.state_snapshot, output: data.output, error: data.error, }, @@ -410,7 +411,7 @@ export const workflowPreviewSlice = createSlice({ nodeTitle: step.nodeTitle, status: step.status, reasoning: step.reasoning, - stateSnapshot: step.stateSnapshot, + stateDelta: step.stateDelta, output: step.output, error: step.error, startedAt: @@ -426,8 +427,8 @@ export const workflowPreviewSlice = createSlice({ }; if (existingIndex !== -1) { - updatedStep.stateSnapshot = - step.stateSnapshot ?? querySteps[existingIndex].stateSnapshot; + updatedStep.stateDelta = + step.stateDelta ?? querySteps[existingIndex].stateDelta; updatedStep.output = step.output ?? querySteps[existingIndex].output; updatedStep.error = step.error ?? querySteps[existingIndex].error; querySteps[existingIndex] = updatedStep; diff --git a/frontend/src/components/ArtifactSidebar.tsx b/frontend/src/components/ArtifactSidebar.tsx index 4c003703..5788bb01 100644 --- a/frontend/src/components/ArtifactSidebar.tsx +++ b/frontend/src/components/ArtifactSidebar.tsx @@ -521,6 +521,7 @@ export default function ArtifactSidebar({
diff --git a/frontend/src/components/artifactViewUtils.ts b/frontend/src/components/artifactViewUtils.ts index cc986a74..5221149e 100644 --- a/frontend/src/components/artifactViewUtils.ts +++ b/frontend/src/components/artifactViewUtils.ts @@ -188,6 +188,12 @@ export function displayFilename( return version?.filename || artifact.title || `artifact-${artifact.id}`; } +// `frame-ancestors` is ignored when delivered via a element (CSP spec), +// and these CSPs live in the srcdoc document's — there is no HTTP +// response to attach a real header to. It only produced a console warning while +// doing nothing; the preview iframe's framing is already constrained by its +// sandbox (no allow-same-origin), so the directive is omitted here. + /** * Restrictive CSP for the preview iframe. No network, no scripts from origins, * no plugins; inline styles/SVG allowed so generated markup renders. Used for @@ -195,8 +201,7 @@ export function displayFilename( */ export const PREVIEW_CSP = "default-src 'none'; img-src data: blob:; style-src 'unsafe-inline'; " + - "font-src data:; script-src 'none'; form-action 'none'; base-uri 'none'; " + - "frame-ancestors 'none'"; + "font-src data:; script-src 'none'; form-action 'none'; base-uri 'none'"; /** * CSP for the mermaid preview. Mermaid's `securityLevel: 'sandbox'` output is an @@ -208,7 +213,7 @@ export const PREVIEW_CSP = export const PREVIEW_CSP_MERMAID = "default-src 'none'; img-src data: blob:; style-src 'unsafe-inline'; " + "font-src data:; script-src 'none'; frame-src data:; form-action 'none'; " + - "base-uri 'none'; frame-ancestors 'none'"; + "base-uri 'none'"; /** * Wrap artifact markup in a minimal HTML document carrying the restrictive CSP. diff --git a/frontend/src/components/ui/sheet.tsx b/frontend/src/components/ui/sheet.tsx index 8ce52880..94485a14 100644 --- a/frontend/src/components/ui/sheet.tsx +++ b/frontend/src/components/ui/sheet.tsx @@ -47,10 +47,15 @@ function SheetContent({ children, side = 'right', showCloseButton = true, + title, ...props }: React.ComponentProps & { side?: 'top' | 'right' | 'bottom' | 'left'; showCloseButton?: boolean; + // Accessible name for the dialog. Radix warns when a Dialog has no Title; + // pass this to render a visually-hidden one when the panel has no visible + // heading of its own (omit it if the children already render a SheetTitle). + title?: string; }) { return ( @@ -71,6 +76,7 @@ function SheetContent({ )} {...props} > + {title ? {title} : null} {children} {showCloseButton && ( diff --git a/frontend/src/conversation/ConversationBubble.tsx b/frontend/src/conversation/ConversationBubble.tsx index 03b025fc..f38495ab 100644 --- a/frontend/src/conversation/ConversationBubble.tsx +++ b/frontend/src/conversation/ConversationBubble.tsx @@ -851,6 +851,7 @@ const ConversationBubble = forwardRef<
diff --git a/tests/agents/test_workflow_engine_coverage.py b/tests/agents/test_workflow_engine_coverage.py index 0aed618b..d6b38761 100644 --- a/tests/agents/test_workflow_engine_coverage.py +++ b/tests/agents/test_workflow_engine_coverage.py @@ -573,7 +573,7 @@ class TestGetExecutionSummary: "started_at": now, "completed_at": now, "error": None, - "state_snapshot": {}, + "state_delta": {}, } ] summary = engine.get_execution_summary() @@ -879,10 +879,10 @@ class TestStepTelemetry: 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"] + assert "query" in by_node["n1"]["state_delta"] # ...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"] == {} + assert by_node["n2"]["state_delta"] == {"x": 2} + assert by_node["n3"]["state_delta"] == {} # The streamed step events carry the same deltas. streamed = { @@ -890,8 +890,8 @@ class TestStepTelemetry: 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"] + assert streamed["n2"]["state_delta"] == {"x": 2} + assert "query" not in streamed["n2"]["state_delta"] @pytest.mark.unit def test_steps_carry_duration_ms(self): @@ -925,7 +925,7 @@ class TestStepTelemetry: 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"] + assert failed and "query" not in failed[0]["state_delta"] failed_log = engine.execution_log[-1] assert failed_log["status"] == "failed" assert failed_log["duration_ms"] >= 0 @@ -962,3 +962,99 @@ class TestStepTelemetry: "status": "completed", }, ] + + +class TestNodeDocumentManifest: + """The engine appends a per-node manifest of staged input documents to the + node system prompt, scoped to that node's input_documents selection.""" + + _ART_1 = "11111111-1111-4111-8111-111111111111" + _ART_2 = "22222222-2222-4222-8222-222222222222" + + def _engine_with_docs(self, monkeypatch): + from contextlib import contextmanager + + engine = WorkflowEngine(_make_graph([], []), _make_agent()) + engine.state["input_documents"] = [ + {"artifact_id": self._ART_1, "ref": "A1", "filename": "portfolio_v1.csv"}, + {"artifact_id": self._ART_2, "ref": "A2", "filename": "notes.txt"}, + ] + + rows = { + self._ART_1: ( + {"current_version": 3, "metadata": {"ref_seq": 1}, "title": "Portfolio"}, + {"filename": "portfolio_v1.csv", "mime_type": "text/csv"}, + ), + self._ART_2: ( + {"current_version": 1, "metadata": {}, "title": "Notes"}, + {"filename": "notes.txt", "mime_type": "text/plain"}, + ), + } + + class FakeRepo: + def __init__(self, conn): + pass + + def resolve_id_by_ref_seq(self, position, conversation_id=None, workflow_run_id=None): + for artifact_id, (artifact, _version) in rows.items(): + if artifact["metadata"].get("ref_seq") == position: + return artifact_id + return None + + def artifact_id_at_position(self, position, conversation_id=None, workflow_run_id=None): + return None + + def get_artifact_in_parent(self, artifact_id, conversation_id=None, workflow_run_id=None): + entry = rows.get(artifact_id) + return entry[0] if entry else None + + def get_version(self, artifact_id, version): + return rows[artifact_id][1] + + @contextmanager + def fake_readonly(): + yield object() + + monkeypatch.setattr("application.storage.db.session.db_readonly", fake_readonly) + monkeypatch.setattr( + "application.storage.db.repositories.artifacts.ArtifactsRepository", FakeRepo + ) + return engine + + @pytest.mark.unit + def test_manifest_lists_refs_filenames_and_mimes(self, monkeypatch): + from application.agents.workflows.schemas import AgentNodeConfig + + engine = self._engine_with_docs(monkeypatch) + config = AgentNodeConfig(agent_type="classic", input_documents=["*"]) + manifest = engine._node_document_manifest(config) + assert "- A1: portfolio_v1.csv (text/csv)" in manifest + # Legacy artifact without ref_seq falls back to the full id, which tools accept. + assert f"- {self._ART_2}: notes.txt (text/plain)" in manifest + assert "Pass a ref" in manifest + + @pytest.mark.unit + def test_manifest_empty_without_selection(self, monkeypatch): + from application.agents.workflows.schemas import AgentNodeConfig + + engine = self._engine_with_docs(monkeypatch) + config = AgentNodeConfig(agent_type="classic", input_documents=[]) + assert engine._node_document_manifest(config) == "" + + @pytest.mark.unit + def test_manifest_failure_never_breaks_the_node(self, monkeypatch): + from contextlib import contextmanager + + from application.agents.workflows.schemas import AgentNodeConfig + + engine = WorkflowEngine(_make_graph([], []), _make_agent()) + engine.state["input_documents"] = [{"artifact_id": self._ART_1}] + + @contextmanager + def broken_readonly(): + raise RuntimeError("db down") + yield + + monkeypatch.setattr("application.storage.db.session.db_readonly", broken_readonly) + config = AgentNodeConfig(agent_type="classic", input_documents=["*"]) + assert engine._node_document_manifest(config) == "" diff --git a/tests/agents/test_workflow_schemas.py b/tests/agents/test_workflow_schemas.py index a4dd96e2..d30a7846 100644 --- a/tests/agents/test_workflow_schemas.py +++ b/tests/agents/test_workflow_schemas.py @@ -468,7 +468,7 @@ class TestNodeExecutionLog: assert log.node_id == "n1" assert log.completed_at is None assert log.error is None - assert log.state_snapshot == {} + assert log.state_delta == {} @pytest.mark.unit def test_full_log(self): @@ -481,10 +481,10 @@ class TestNodeExecutionLog: started_at=started, completed_at=completed, error=None, - state_snapshot={"key": "value"}, + state_delta={"key": "value"}, ) assert log.completed_at == completed - assert log.state_snapshot == {"key": "value"} + assert log.state_delta == {"key": "value"} @pytest.mark.unit def test_extra_forbidden(self):