From 5dff7cdfd3ebe675f1be5ace0579774e8687ac23 Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Thu, 13 Aug 2026 21:50:49 +0530 Subject: [PATCH 1/2] fix(workflows): an approval must not repeat an outward call, finish a run that is waiting, or hide what it is approving (#846) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three defects on the workflow approval path. **A continuation repeated an outward call.** #438's requirement was never "stop re-executing" — a paused tinyflows run is settled, so approving is a re-run — it was that an approval must never send twice. #496 met that for the half the host routes itself (an `output` node's report) and could not reach the half the engine performs mid-run. A new outward-call ledger rides the approval card and the continuation input like #496's, and the node is rewritten before the run starts to replay its recorded result rather than call again. **A run reported Finished with its approval outstanding.** The console read only parked *deliveries*, so a run stopped at a gate — which never reaches an output node, and so has none — fell through every branch to "Finished". **The card never said what it was approving.** A call description existed only for policy-raised gates, so on a `full`-tier company every workflow card named a node id and showed the engine's resume payload. The host now describes an authored gate too, and the console labels the card by its tool. --- frontend/src/lib/language.ts | 102 ++- .../src/views/workflows/RunHistoryPanel.tsx | 40 +- frontend/src/views/workflows/run-health.ts | 27 +- frontend/test/unit/workflow-gate-card.test.ts | 168 +++++ src/runtime/workflow_resume.rs | 423 ++++++++++- src/workflows/caps/dry_run.rs | 13 + src/workflows/caps/tools.rs | 64 ++ src/workflows/gate.rs | 202 ++++- src/workflows/mod.rs | 3 + src/workflows/replay.rs | 701 ++++++++++++++++++ src/workflows/runner.rs | 100 ++- 11 files changed, 1783 insertions(+), 60 deletions(-) create mode 100644 frontend/test/unit/workflow-gate-card.test.ts create mode 100644 src/workflows/replay.rs diff --git a/frontend/src/lib/language.ts b/frontend/src/lib/language.ts index 84051dd6a..b0184af8c 100644 --- a/frontend/src/lib/language.ts +++ b/frontend/src/lib/language.ts @@ -266,6 +266,20 @@ const TOOL_LABELS: Readonly> = { * left alone so no other surface shifts under this change. */ export function approvalAction(a: ApprovalSummary): string { + // Issue #846: a paused workflow gate is named by the CALL it is stopping, not + // by the mechanism that stopped it. "Continue a paused workflow" is true of + // every one of these cards and therefore tells an operator nothing — it is + // #372's "Glob" complaint, one surface over: the chat path was fixed by #375 + // and this path was not, because its label came from `EFFECT_LABELS` (which + // keys on the effect kind, and the kind here is always `workflow.approve`) + // rather than from the tool. + // + // The tool is on the wire whenever the host could classify the node's call. + // When it could not — an authored gate on a step that calls nothing — the + // glossary entry below is still the honest answer, so this promotes and never + // hides. + const workflowTool = workflowGateTool(a); + if (workflowTool) return toolAction(workflowTool); return ( labelFor(EFFECT_LABELS, a.kind) ?? labelFor(TOOL_LABELS, a.kind) ?? @@ -273,6 +287,29 @@ export function approvalAction(a: ApprovalSummary): string { ); } +/** The effect kind a paused workflow gate parks as — mirrors + * `WORKFLOW_APPROVE_KIND` in `src/runtime/workflow_resume.rs`. */ +const WORKFLOW_APPROVE_KIND = "workflow.approve"; + +/** + * The tool a paused workflow gate is stopping, when the host named one + * (issue #846) — `null` for every other kind, and for a gate whose node makes + * no classifiable call. + * + * Reads `payload.tool`, which the host writes for a policy-raised gate (#460) + * and, since #846, for an authored one too. Absence is meaningful and is + * preserved as such: an **older host** omits the key entirely, and this then + * returns `null` so the card renders exactly the pre-#846 line rather than an + * empty label. + */ +function workflowGateTool(a: ApprovalSummary): string | null { + if (a.kind !== WORKFLOW_APPROVE_KIND) return null; + const payload = a.payload; + if (payload == null || typeof payload !== "object" || Array.isArray(payload)) return null; + const tool = (payload as Record).tool; + return typeof tool === "string" && tool !== "" ? tool : null; +} + /** * What a tool does, in plain language, from its identifier alone (#374). * @@ -324,7 +361,10 @@ export function payloadLines(a: ApprovalSummary): PayloadLine[] { if (typeof payload !== "object") return [{ label: "value", value: renderValue(payload) }]; if (Array.isArray(payload)) return [{ label: "items", value: renderValue(payload) }]; - const entries = Object.entries(payload as Record); + const entries = + a.kind === WORKFLOW_APPROVE_KIND + ? workflowGateEntries(payload as Record) + : Object.entries(payload as Record); if (entries.length === 0) return []; // Preferred ordering for the kinds whose argument names we know. Unlisted @@ -340,6 +380,66 @@ export function payloadLines(a: ApprovalSummary): PayloadLine[] { .map(([label, value]) => ({ label, value: renderValue(value) })); } +/** + * The payload of a paused workflow gate, as the lines an operator needs + * (issue #846). + * + * A `workflow.approve` payload is not a tool call's arguments — it is the host's + * **resume record**, and every other card's rule ("show the payload, it is what + * you are consenting to") produces exactly the wrong thing here. What an + * operator saw was `input: {"items":[{"json":{}}],"port":null}`: the engine's + * seed payload, verbatim, as the sole description of a decision. + * + * So this promotes the call's own arguments to the top level, where every other + * card carries them, and drops the machinery. The dropped keys are dropped for + * stated reasons, not for tidiness: + * + * * `input` — the resume payload. It is engine seed data, it is the thing that + * read as a description and was not one, and it also carries an `approvals` + * list that accumulates down a lineage: an operator seeing + * `{"approvals":["fetch_bbc","fetch_espn"]}` reads it as a card covering all + * of them, which it is not. (Consolidating several gates into one card is + * real and is issue #842 — not something to imply by accident here.) + * * `delivered`, `performed` — the #438/#846 ledgers. They say what will NOT + * happen again, which is a property of the mechanism and is already stated in + * prose by `note`. + * * `content` — rendered in full by its own block (`WorkflowContentReview`, + * #596); repeating it as one clamped JSON line would be strictly worse. + * * `note` — prose, rendered as prose elsewhere on the card, not as a + * `key: value` pair in a monospace block. + * + * Everything else survives, including any key a **newer host** adds that this + * console has never heard of. Dropping unknown keys would make an old console + * silently hide new information, which is the failure mode this whole file is + * written against; the denylist is closed and the allowlist is not. + */ +function workflowGateEntries(payload: Record): [string, unknown][] { + const args = payload.args; + const argEntries: [string, unknown][] = + args != null && typeof args === "object" && !Array.isArray(args) + ? Object.entries(args as Record) + : []; + const rest = Object.entries(payload).filter(([key]) => !WORKFLOW_GATE_HIDDEN.has(key)); + // The call first, then where and what stopped it. An operator decides on the + // call; the node id is how they find it afterwards. This order survives + // `payloadLines`' sort because no `PAYLOAD_KEY_ORDER` entry exists for this + // kind — every rank is equal, and the sort is stable. + return [...argEntries, ...rest]; +} + +/** Payload keys of a `workflow.approve` card that are machinery, not the + * decision — see {@link workflowGateEntries} for why each one is here. */ +const WORKFLOW_GATE_HIDDEN: ReadonlySet = new Set([ + "input", + "delivered", + "performed", + "content", + "note", + // Unwrapped one level above, so keeping it would print the same arguments + // twice — once readably and once as a JSON blob. + "args", +]); + /** Which arguments lead the preview, per tool. Presentation only. */ const PAYLOAD_KEY_ORDER: Readonly> = { shell: ["command", "cwd", "timeout"], diff --git a/frontend/src/views/workflows/RunHistoryPanel.tsx b/frontend/src/views/workflows/RunHistoryPanel.tsx index d089fcbc1..a6d1e953a 100644 --- a/frontend/src/views/workflows/RunHistoryPanel.tsx +++ b/frontend/src/views/workflows/RunHistoryPanel.tsx @@ -16,7 +16,7 @@ import type { } from "@/api/workflows"; import { failedNodeOf } from "./graph"; -import { pendingCount, relativeTime, runTone, undeliveredCount } from "./run-health"; +import { awaitingCount, relativeTime, runTone, undeliveredCount } from "./run-health"; /** Badge styling per delivery outcome. A report that did NOT go out must not * look like one that did — `denied` and `failed` are the two an operator has to @@ -82,7 +82,9 @@ export function DeliveryRows({ deliveries }: { deliveries: DeliveryReport[] }) { export function LastRunChip({ run }: { run: WorkflowRunOutcome }) { const tone = runTone(run); const undelivered = undeliveredCount(run.deliveries); - const pending = pendingCount(run.deliveries); + // Issue #846: gates and parked reports together. The chip said "Manual run" + // and a green dot for a run whose first node was still waiting on a person. + const awaiting = awaitingCount(run); return ( 0 ? ` · ${undelivered} not delivered` - : pending > 0 - ? ` · ${pending} awaiting approval` + : awaiting > 0 + ? ` · ${awaiting} awaiting approval` : ""} · {relativeTime(run.atMillis)} @@ -291,6 +293,36 @@ function RunHistoryRow({

Still running — reports are routed when it finishes.

+ ) : run.pendingApprovals.length > 0 ? ( + <> + {/* A paused run can still have routed reports — the output nodes it + reached BEFORE the gate. Those rows are shown as they always were, + with the waiting line above rather than instead of them: replacing + them would trade one silent omission for another. */} + {run.deliveries.length > 0 && } + // Issue #846. This is the arm that was missing, and its absence is how a + // run waiting on a human came to report success: a paused run has no + // error, no cancellation, is not `running` (the engine settled it) and + // routed nothing, so it fell through every branch to the "Finished" line + // below — while its gate sat undecided on the Approvals page. + // + // "Not finished" is the claim, stated in the operator's terms rather + // than the engine's. The run object really is settled; what has not + // happened is the work, and the work is what the operator is asking + // about. Naming the nodes matters as much as the state: a scheduled run + // that silently did nothing is exactly the failure this reads as, and + // the fix is a click, so the row says which click. +

+ Not finished — waiting for your approval on{" "} + {run.pendingApprovals.map((node) => `“${node}”`).join(", ")}. Nothing past{" "} + {run.pendingApprovals.length === 1 ? "it" : "them"} has run + {run.deliveries.length === 0 ? ", and no reports were routed" : ""}. Approve or decline + it in Approvals to carry the run on. +

+ ) : run.deliveries.length > 0 ? ( // Deliberately the SAME component the live run drawer uses, so a report // reads identically whether it's on screen now or a week old. diff --git a/frontend/src/views/workflows/run-health.ts b/frontend/src/views/workflows/run-health.ts index 41be876a0..f6d0f35e2 100644 --- a/frontend/src/views/workflows/run-health.ts +++ b/frontend/src/views/workflows/run-health.ts @@ -33,6 +33,25 @@ export function pendingCount(deliveries: DeliveryReport[]): number { return deliveries.filter((d) => d.status === PENDING_STATUS).length; } +/** + * Everything about this run that is waiting on a person: the gates it paused at + * **and** the reports it parked (issue #846). + * + * The two were never read together, and that is what let a run report success + * while a human had not answered it. `pendingCount` sees only `deliveries`, so a + * run that paused at a `requires_approval` node and therefore never reached an + * `output` node at all — the exact shape of a gated workflow — has an empty + * `deliveries` array and scored as a clean run. + * + * `pendingApprovals` has been on the wire since #395 and the history row has + * badged its count since; nothing read it for the run's *state*. This is that + * read, in one place, so the tone, the chip and the row cannot disagree about + * whether somebody is being waited on. + */ +export function awaitingCount(run: WorkflowRunOutcome): number { + return (run.pendingApprovals?.length ?? 0) + pendingCount(run.deliveries); +} + /** A compact "N minutes ago" for a run timestamp — enough to tell last night's * scheduled run from the one just clicked, without a date library. */ export function relativeTime(atMillis: number): string { @@ -77,8 +96,12 @@ export function runTone(run: WorkflowRunOutcome): { dot: string; label: string } // Blocked, not running. This was the running colour, which said "the machine // is working on it" about the one state that means the opposite: it is // parked until a human decides. Amber is the colour that gets looked at. - if (pendingCount(run.deliveries) > 0) - return { dot: "bg-status-blocked", label: "awaiting approval" }; + // + // Issue #846: `awaitingCount`, not `pendingCount`. A run that paused at a gate + // parked no report — it never reached an output node — so the delivery-only + // read scored it green and the operator was told a run that did none of its + // work had succeeded. + if (awaitingCount(run) > 0) return { dot: "bg-status-blocked", label: "awaiting approval" }; return { dot: "bg-status-done", label: "ok" }; } diff --git a/frontend/test/unit/workflow-gate-card.test.ts b/frontend/test/unit/workflow-gate-card.test.ts new file mode 100644 index 000000000..6fc4efdee --- /dev/null +++ b/frontend/test/unit/workflow-gate-card.test.ts @@ -0,0 +1,168 @@ +import { describe, expect, it } from "vitest"; + +import { approvalAction, payloadLines } from "@/lib/language"; +import { awaitingCount, runTone } from "@/views/workflows/run-health"; +import type { ApprovalSummary } from "@/api/types"; +import type { WorkflowRunOutcome } from "@/api/workflows"; + +/** + * Issue #846, the two console-side halves. + * + * **The card** said "Continue a paused workflow" and showed the engine's resume + * payload — a label true of every one of these cards, over a description that + * describes nothing. #372 made exactly this complaint about the chat surface and + * #375 fixed it there; this is the same treatment for the workflow-originated + * card, now that the host names the call on it. + * + * **The run** said "Finished — this run routed no reports" while its gate sat + * undecided on the Approvals page. A run waiting on a person is not finished, + * and a scheduled run that reports success while doing none of its work is the + * failure that makes it matter. + */ + +function approval(over: Partial & Pick): ApprovalSummary { + return { id: "a1", amount_usd: null, at_millis: 1_000, agent: null, ...over }; +} + +/** A paused gate's card as the host now parks it. */ +function gateCard(payload: Record): ApprovalSummary { + return approval({ kind: "workflow.approve", payload }); +} + +/** The card the reproduction in #846 actually produced. */ +const AS_REPORTED = { + workflow_id: "daily-sports-news-blog", + node_id: "fetch_bbc", + input: { items: [{ json: {} }], port: null }, +}; + +/** The same gate, as the host describes it after this change. */ +const DESCRIBED = { + ...AS_REPORTED, + tool: "web_fetch", + args: { url: "https://www.bbc.com/sport" }, + target: "www.bbc.com", + delivered: [], + performed: [], + note: "Approving this re-runs the whole workflow from the start…", +}; + +function run(over: Partial = {}): WorkflowRunOutcome { + return { + seq: 1, + atMillis: 1_000, + scheduled: false, + running: false, + cancelled: false, + error: null, + deliveries: [], + pendingApprovals: [], + nodes: [], + ...over, + } as WorkflowRunOutcome; +} + +describe("a paused workflow gate says what it is approving (#846)", () => { + it("names the tool rather than the mechanism", () => { + expect(approvalAction(gateCard(DESCRIBED))).toBe("Fetch a web page"); + }); + + it("shows the call's arguments and its destination, not the resume payload", () => { + const lines = payloadLines(gateCard(DESCRIBED)); + const byLabel = Object.fromEntries(lines.map((l) => [l.label, l.value])); + + // What the operator is deciding about — the complaint in the issue was that + // neither the tool nor the host appeared anywhere on this card. + expect(byLabel.url).toBe("https://www.bbc.com/sport"); + expect(byLabel.target).toBe("www.bbc.com"); + // Where to find it afterwards. + expect(byLabel.node_id).toBe("fetch_bbc"); + expect(byLabel.workflow_id).toBe("daily-sports-news-blog"); + + // The arguments lead. An operator scanning a clamped block must see the + // call before the bookkeeping. + expect(lines[0].label).toBe("url"); + }); + + it("hides the engine's resume payload and the mechanism's ledgers", () => { + const labels = payloadLines(gateCard(DESCRIBED)).map((l) => l.label); + + // `input` is the seed payload that READ as a description and was not one. + // It also carries the accumulating `approvals` list, which an operator + // reasonably reads as one card covering several gates — it is not, and + // consolidating them is #842's job, not something to imply by accident. + expect(labels).not.toContain("input"); + expect(labels).not.toContain("delivered"); + expect(labels).not.toContain("performed"); + expect(labels).not.toContain("note"); + // Unwrapped one level up, so it must not also appear as a JSON blob. + expect(labels).not.toContain("args"); + }); + + it("keeps a key a newer host adds that this console has never heard of", () => { + const labels = payloadLines(gateCard({ ...DESCRIBED, whodunnit: "a future field" })).map( + (l) => l.label, + ); + expect(labels).toContain("whodunnit"); + }); + + it("falls back to the old line for an older host that names no tool", () => { + // Absence is meaningful: a host from before this change omits `tool`, and + // the card must read exactly as it did rather than showing a blank label. + expect(approvalAction(gateCard(AS_REPORTED))).toBe("Continue a paused workflow"); + }); + + it("leaves every other kind's card alone", () => { + const shell = approval({ + kind: "shell", + agent: "ceo", + payload: { command: "ls -la", cwd: "/tmp" }, + }); + expect(approvalAction(shell)).toBe("Run a terminal command"); + expect(payloadLines(shell).map((l) => l.label)).toEqual(["command", "cwd"]); + }); +}); + +describe("a run waiting on a person is not finished (#846)", () => { + it("counts a parked gate as awaiting, not as a clean run", () => { + const parked = run({ pendingApprovals: ["fetch_bbc"] }); + + // The reproduction's exact shape: no error, not cancelled, not running, + // and no deliveries — because it never reached an output node. Every one of + // those reads as "fine" on its own, which is how it scored green. + expect(awaitingCount(parked)).toBe(1); + expect(runTone(parked).label).toBe("awaiting approval"); + }); + + it("still reads as ok when nothing is waiting", () => { + expect(runTone(run()).label).toBe("ok"); + expect(awaitingCount(run())).toBe(0); + }); + + it("counts parked gates and parked reports together", () => { + const both = run({ + pendingApprovals: ["gate_a", "gate_b"], + deliveries: [ + { + node: "summary", + kind: "owner", + status: "pending", + detail: "waiting", + target: null, + reason: null, + }, + ], + } as Partial); + expect(awaitingCount(both)).toBe(3); + }); + + it("does not let a still-running run be read as awaiting", () => { + // `runTone` checks in-flight first, and it must keep doing so: a run that + // has not finished has neither succeeded nor stopped for a human. + expect(runTone(run({ running: true, pendingApprovals: ["gate"] })).label).toBe("running"); + }); + + it("keeps a failure louder than a pending approval", () => { + expect(runTone(run({ error: "boom", pendingApprovals: ["gate"] })).label).toBe("failed"); + }); +}); diff --git a/src/runtime/workflow_resume.rs b/src/runtime/workflow_resume.rs index 5a1289277..5b8e7579a 100644 --- a/src/runtime/workflow_resume.rs +++ b/src/runtime/workflow_resume.rs @@ -139,6 +139,15 @@ pub const PAYLOAD_INPUT: &str = "input"; /// The payload key holding this lineage's delivery ledger (issue #438) — the /// reports a continuation must NOT send again. pub const PAYLOAD_DELIVERED: &str = "delivered"; +/// The payload key holding this lineage's **outward-call** ledger (issue #846) +/// — the `tool_call` nodes that already reached a counterparty, and the result +/// each returned, so a continuation replays them instead of calling again. +/// +/// The `output`-node sibling of [`PAYLOAD_DELIVERED`], for the other half of +/// #438's exposure: #496 guarded a report a *delivery* node routed, and nothing +/// guarded a `send` / `publish` / `repo_publish` the graph made as a node of its +/// own. See [`PerformedCall`]. +pub const PAYLOAD_PERFORMED: &str = "performed"; /// The payload key holding the plain-prose statement of what approving costs. pub const PAYLOAD_NOTE: &str = "note"; /// The payload key holding the verbatim output of the gate's upstream nodes — @@ -155,24 +164,45 @@ pub const PAYLOAD_TOOL: &str = "tool"; /// (issue #460) — the same sentence the agent path puts on its card. Absent for /// the same reason as [`PAYLOAD_TOOL`]. pub const PAYLOAD_REASON: &str = "reason"; -/// The payload key holding what a policy-gated call would reach — `"POST -/// api.example.com"` for an `http_request` node (issue #614). Absent when the +/// The payload key holding what a gated call would reach — `"POST +/// api.example.com"` for an `http_request` node (issue #614), the host for a +/// `tool_call` node whose arguments name a URL (issue #846). Absent when the /// node's call has no destination worth naming, or when the URL is still an /// unresolved `=`-expression at gate time. pub const PAYLOAD_TARGET: &str = "target"; +/// The payload key holding the gated node's **authored arguments** (issue #846) +/// — the `url` a `web_fetch` will fetch, the recipient a `send` will reach. +/// +/// This is what makes a workflow card decidable in the way #372/#375 made a +/// chat card decidable: the operator sees the call, not a node id. It is +/// credential-redacted host-side by the same projection that redacts a tool +/// call's payload on the chat path — see +/// [`display_payload`](crate::runtime::approval_display) — so this key adds no +/// new redaction rule and cannot bypass the existing one. +/// +/// Absent when the node makes no classifiable call (an authored gate on a +/// `transform`, say), which the console must render as "no arguments" rather +/// than as an empty object. +pub const PAYLOAD_ARGS: &str = "args"; -/// What a policy-gated node's card says about the call being decided -/// (issues #460, #614). +/// What a gated node's card says about the call being decided (issues #460, +/// #614, #846). /// -/// Grouped rather than passed as three more arguments to [`gate_effect`]: they -/// are written together or not at all, and an authored `requires_approval` gate -/// passes `None` for the lot — no particular call is being decided there. +/// Grouped rather than passed as four more arguments to [`gate_effect`]: they +/// describe one thing — the call — and a node whose call the host cannot +/// classify at all passes `None` for the lot. #[derive(Debug, Clone, Copy)] pub struct GateCall<'a> { /// The tool the node would run. pub tool: &'a str, - /// The policy's own words for why it stopped. - pub reason: &'a str, + /// The policy's own words for why it stopped — `None` on an authored + /// `requires_approval` gate, where nobody wrote a reason because nobody was + /// asked to. The call is still named (issue #846). + pub reason: Option<&'a str>, + /// The node's authored arguments, so the operator decides about a call + /// rather than about a node id (issue #846). Redacted downstream, at the + /// same projection that redacts a chat card's payload. + pub args: Option<&'a Value>, /// Method and host, when knowable. Never the path or query — see /// `GatedCall::target` in `crate::workflows::gate` for why. pub target: Option<&'a str>, @@ -188,6 +218,17 @@ pub struct GateCall<'a> { /// every continuation gate a "new" decision and stack a duplicate card. pub const CONTINUATION_DELIVERED_KEY: &str = "__opencompany_delivered"; +/// The reserved trigger-input key the **outward-call** ledger rides into a +/// continuation run under (issue #846). +/// +/// Reserved on exactly the terms [`CONTINUATION_DELIVERED_KEY`] is: host-written, +/// host-read, never authored and never seen by the engine as anything but +/// opaque trigger data. Stripped before two parked gates are compared +/// ([`is_same_gate`]) for the same reason its sibling is — it describes what has +/// already happened, not what is being decided, so counting it would make every +/// continuation gate read as a new decision and stack a duplicate card. +pub const CONTINUATION_PERFORMED_KEY: &str = "__opencompany_performed"; + /// What approving a workflow gate actually does, in the operator's own terms. /// /// This rides the card as [`PAYLOAD_NOTE`] rather than living only in a design @@ -197,7 +238,8 @@ pub const CONTINUATION_DELIVERED_KEY: &str = "__opencompany_delivered"; /// Approvals card. pub const CONTINUATION_NOTE: &str = "Approving this re-runs the whole workflow from the start — every step before this gate runs \ again, and any agent steps spend tokens again. Reports this run already delivered will not be \ - sent a second time."; + sent a second time, and a step that already sent or published something replays what it \ + returned instead of doing it again."; /// One `output` node whose report a run in this lineage has already delivered. /// @@ -214,6 +256,74 @@ pub struct DeliveredReport { pub kind: String, } +/// One `tool_call` node whose call **left the building** in a prior run of this +/// lineage, together with the result it returned (issue #846). +/// +/// # Why this exists beside [`DeliveredReport`] rather than inside it +/// +/// #438's exposure is "approving re-runs the graph, so something that already +/// left the building leaves it again". #496 closed that for the half the host +/// performs itself — an `output` node's report, routed by `deliver_outputs` — +/// because that is the half the host can skip by simply not calling out. A +/// `tool_call` node's send is performed by the **engine**, through a capability, +/// and the host cannot decline it after the fact: it has to arrange, before the +/// run starts, for the call not to be made. So the identity is the same (the +/// node) but the mechanism is not, and folding the two ledgers into one type +/// would put a `result` field on a record whose whole point is that there is +/// nothing to replay. +/// +/// `result` is the **verbatim capability return** — the value the engine wrapped +/// in its `{ json, text, raw }` envelope — so replaying it reconstructs the +/// node's output byte-for-byte rather than approximating it. A node whose +/// recorded result would have to be truncated to fit the card is deliberately +/// **not** recorded: see `outward_calls_performed` in `crate::workflows::replay`. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct PerformedCall { + /// The `tool_call` node that made the call. + pub node: String, + /// The toolbelt slug it invoked, for the log line and the operator's card. + pub tool: String, + /// The verbatim value the capability returned. + pub result: Value, +} + +/// The outward calls this lineage has already made, read off a trigger input. +/// +/// Tolerant on exactly [`delivered_in_input`]'s terms, and for the same reason: +/// a missing key, a non-array or a malformed row yields "nothing known to have +/// been performed", which is the pre-#846 behaviour (call it). The failure mode +/// of being wrong in the other direction is a node that silently never runs. +pub fn performed_in_input(input: &Value) -> Vec { + input + .get(CONTINUATION_PERFORMED_KEY) + .and_then(Value::as_array) + .map(|rows| { + rows.iter() + .filter_map(|row| serde_json::from_value(row.clone()).ok()) + .collect() + }) + .unwrap_or_default() +} + +/// The outward-call ledger a gate parked on this run should carry: what this run +/// performed, unioned with what its own trigger input already listed. +/// +/// The union is what makes a **two-gate** graph correct, exactly as +/// [`delivery_ledger`]'s is: a continuation that replayed a send rather than +/// making it has performed nothing itself, so a ledger built from this run alone +/// would be empty at the second gate and the third run would send for real. +/// First entry per node wins — the earliest run in the lineage is the one that +/// actually reached the counterparty, and its result is the one to replay. +fn performed_ledger(input: &Value, performed: &[PerformedCall]) -> Vec { + let mut ledger = performed_in_input(input); + for call in performed { + if !ledger.iter().any(|prior| prior.node == call.node) { + ledger.push(call.clone()); + } + } + ledger +} + /// The reports this lineage has already delivered, read off a trigger input. /// /// Tolerant by construction — a missing key, a non-array, or a malformed row @@ -291,18 +401,25 @@ fn delivery_ledger(input: &Value, deliveries: &[DeliveryReport]) -> Vec>, ) -> Effect { let mut payload = Map::new(); @@ -317,15 +434,39 @@ pub fn gate_effect( PAYLOAD_DELIVERED.to_string(), json!(delivery_ledger(input, deliveries)), ); + // Issue #846: what must NOT be *called* again when this card is approved. + // Written unconditionally, including when empty, so a reader can tell a host + // that considered the question and found nothing from one that never asked. + payload.insert( + PAYLOAD_PERFORMED.to_string(), + json!(performed_ledger(input, performed)), + ); // What approving costs, in the operator's own terms. payload.insert(PAYLOAD_NOTE.to_string(), json!(CONTINUATION_NOTE)); // Issue #460: which call the policy stopped, and why. The keys are ABSENT // rather than null on an authored gate — a card that names no tool is a // different thing from one whose tool could not be determined, and a // console reading `payload.tool` should be able to tell them apart. + // Issue #460: which call was stopped, and — when a policy stopped it — why. + // The keys are ABSENT rather than null on a node whose call cannot be + // classified: a card that names no tool is a different thing from one whose + // tool could not be determined, and a console reading `payload.tool` should + // be able to tell them apart. + // + // Issue #846: `reason` remains policy-only while `tool` / `args` / `target` + // are written for an **authored** gate too. That asymmetry is the point. An + // author's `requires_approval` has no reason to state — they asked for a + // human, and the console says so in its own words — but the call itself is + // just as knowable, and a card that withholds it because *nobody wrote a + // sentence about it* is the bug this closes. if let Some(call) = call { payload.insert(PAYLOAD_TOOL.to_string(), json!(call.tool)); - payload.insert(PAYLOAD_REASON.to_string(), json!(call.reason)); + if let Some(reason) = call.reason { + payload.insert(PAYLOAD_REASON.to_string(), json!(reason)); + } + if let Some(args) = call.args { + payload.insert(PAYLOAD_ARGS.to_string(), args.clone()); + } if let Some(target) = call.target { payload.insert(PAYLOAD_TARGET.to_string(), json!(target)); } @@ -440,11 +581,16 @@ fn decided_input(effect: &Effect) -> Option { .map(without_ledger) } -/// `input` with the reserved delivery-ledger key removed. A non-object input is -/// returned as-is — there is nothing to strip. +/// `input` with the reserved host-threaded ledger keys removed. A non-object +/// input is returned as-is — there is nothing to strip. +/// +/// Both keys, and neither is optional: a continuation's input differs from the +/// paused run's by exactly these, so letting either difference count would make +/// every continuation gate a "new" decision and stack a duplicate card. fn without_ledger(mut input: Value) -> Value { if let Value::Object(map) = &mut input { map.remove(CONTINUATION_DELIVERED_KEY); + map.remove(CONTINUATION_PERFORMED_KEY); } input } @@ -554,7 +700,52 @@ pub(crate) fn continuation_input(effect: &Effect) -> Result { .collect() }) .unwrap_or_default(); - Ok(with_delivered(with_approval(input, node_id), &delivered)) + // Issue #846: the outward-call ledger travels on the same terms and for the + // same reason. An input that carries the approval but not this resumes and + // **re-sends**, which is #438 on the node the host does not route itself. + let performed: Vec = effect + .payload + .get(PAYLOAD_PERFORMED) + .and_then(Value::as_array) + .map(|rows| { + rows.iter() + .filter_map(|row| serde_json::from_value(row.clone()).ok()) + .collect() + }) + .unwrap_or_default(); + Ok(with_performed( + with_delivered(with_approval(input, node_id), &delivered), + &performed, + )) +} + +/// Writes `performed` onto the trigger input under +/// [`CONTINUATION_PERFORMED_KEY`], replacing whatever was there. +/// +/// Replace rather than merge, on [`with_delivered`]'s reasoning exactly: the +/// card's ledger was *built* by unioning the input's own list with what the run +/// performed ([`performed_ledger`]), so it is already the superset, and merging +/// again here would be a second place for that rule to drift. +/// +/// An empty ledger writes nothing at all, so a first run's input shape is +/// untouched and the reserved key appears only once there is something to +/// suppress. +fn with_performed(input: Value, performed: &[PerformedCall]) -> Value { + if performed.is_empty() { + return input; + } + match input { + Value::Object(mut map) => { + map.insert( + CONTINUATION_PERFORMED_KEY.to_string(), + serde_json::json!(performed), + ); + Value::Object(map) + } + // `with_approval` always yields an object, so this is unreachable + // through `continuation_input`. Kept total rather than panicking. + other => other, + } } /// Writes `delivered` onto the trigger input under @@ -646,7 +837,7 @@ mod tests { use super::*; fn effect(workflow: &str, node: &str, input: Value) -> Effect { - gate_effect(workflow, node, &input, "run-1", &[], None) + gate_effect(workflow, node, &input, "run-1", &[], &[], None) } /// A delivery row with `status`, as `deliver_outputs` would have returned it. @@ -778,6 +969,7 @@ mod tests { delivery("owner_summary", "owner", DeliveryStatus::Sent), delivery("cold_note", "email", DeliveryStatus::Pending), ], + &[], None, ); assert_eq!( @@ -811,6 +1003,7 @@ mod tests { &Value::Null, "run-1", &[delivery("summary", "owner", status)], + &[], None, ); assert!( @@ -833,6 +1026,7 @@ mod tests { delivery("summary", "owner", DeliveryStatus::Sent), delivery("summary", "owner", DeliveryStatus::Sent), ], + &[], None, ); assert_eq!(ledger(&e).len(), 1); @@ -848,6 +1042,7 @@ mod tests { &serde_json::json!({ "request": "x" }), "run-1", &[delivery("summary", "owner", DeliveryStatus::Sent)], + &[], None, ); @@ -881,12 +1076,13 @@ mod tests { &serde_json::json!({ "request": "x" }), "run-1", &[delivery("summary", "owner", DeliveryStatus::Sent)], + &[], None, ); let continuation = continuation_input(&first).expect("continues"); // Run 2 skips the summary (delivering nothing) and pauses on gate-b. - let second = gate_effect("digest", "gate-b", &continuation, "run-2", &[], None); + let second = gate_effect("digest", "gate-b", &continuation, "run-2", &[], &[], None); assert_eq!( ledger(&second), vec![DeliveredReport { @@ -902,6 +1098,168 @@ mod tests { assert_eq!(delivered_in_input(&next).len(), 1); } + // --- issue #846: the outward-call ledger ------------------------------- + + /// One outward call, made once, across a whole lineage — **the headline + /// claim for issue #846**, in the shape #496 proved for the delivery half. + /// + /// A `POST` upstream of two gates. Run 1 fires it and pauses; approving + /// starts run 2, which must NOT fire it again and must pause on the second + /// gate; approving that starts run 3, which must not fire it either. The + /// ledger has to accumulate down the lineage or run 3 posts for real — + /// exactly the trap the delivery ledger's own two-gate test exists for. + /// + /// Asserted on the ledger the card carries and the input it produces, + /// because those are what decide whether the call is made: the graph rewrite + /// that consumes them is pinned in `crate::workflows::replay`, and the + /// invoker arm that answers it is pinned in `crate::workflows::caps::tools`. + #[test] + fn an_outward_call_is_made_once_across_two_gates() { + let posted = PerformedCall { + node: "notify".into(), + tool: "http_request POST".into(), + result: serde_json::json!({ "status": 201 }), + }; + + // Run 1: the POST fires, the run pauses on gate-a. + let first = gate_effect( + "digest", + "gate-a", + &serde_json::json!({ "request": "x" }), + "run-1", + &[], + std::slice::from_ref(&posted), + None, + ); + let continuation = continuation_input(&first).expect("continues"); + assert_eq!( + performed_in_input(&continuation), + vec![posted.clone()], + "the continuation must know what run 1 already posted" + ); + + // Run 2: the first POST is replayed (this run does not repeat it), and + // a SECOND outward node fires before the run pauses on gate-b. + // + // The second call is what makes this the real two-gate case rather than + // a walk-through. A card carrying a non-empty ledger REPLACES the input's + // key rather than merging into it — `with_performed` documents why — so + // if `performed_ledger` did not union the input's own entries first, run + // 2's card would carry only its own call and run 3 would post the first + // one a second time. Reverting that union fails this test on the + // `notify` entry alone. + let also_posted = PerformedCall { + node: "escalate".into(), + tool: "http_request POST".into(), + result: serde_json::json!({ "status": 202 }), + }; + let second = gate_effect( + "digest", + "gate-b", + &continuation, + "run-2", + &[], + std::slice::from_ref(&also_posted), + None, + ); + let next = continuation_input(&second).expect("continues"); + + assert_eq!( + performed_in_input(&next), + vec![posted, also_posted], + "run 3 must be told about BOTH earlier posts, not just the last one" + ); + assert_eq!(next["approvals"], serde_json::json!(["gate-a", "gate-b"])); + } + + /// The earliest result in the lineage wins, and a node is listed once. + /// + /// The run that actually reached the counterparty is the one whose receipt + /// downstream nodes saw, so a later run must not overwrite it — and a ledger + /// that grew an entry per hop would bloat every card in a long lineage. + #[test] + fn the_outward_ledger_keeps_the_first_result_per_node() { + let original = PerformedCall { + node: "notify".into(), + tool: "http_request POST".into(), + result: serde_json::json!({ "id": "first" }), + }; + let input = serde_json::json!({ CONTINUATION_PERFORMED_KEY: [original.clone()] }); + + let card = gate_effect( + "digest", + "gate", + &input, + "run-2", + &[], + &[PerformedCall { + node: "notify".into(), + tool: "http_request POST".into(), + result: serde_json::json!({ "id": "second" }), + }], + None, + ); + + let ledger: Vec = serde_json::from_value( + card.payload + .get(PAYLOAD_PERFORMED) + .expect("carried") + .clone(), + ) + .expect("well-formed"); + assert_eq!(ledger, vec![original]); + } + + /// A card with no outward ledger produces an input with no reserved key — + /// so a first run's trigger payload keeps exactly the shape it always had. + #[test] + fn an_empty_outward_ledger_leaves_the_input_untouched() { + let card = gate_effect( + "digest", + "gate", + &serde_json::json!({ "request": "x" }), + "run-1", + &[], + &[], + None, + ); + let input = continuation_input(&card).expect("continues"); + assert!( + input.get(CONTINUATION_PERFORMED_KEY).is_none(), + "nothing to suppress must write nothing: {input}" + ); + assert!(performed_in_input(&input).is_empty()); + } + + /// The outward ledger is NOT part of a gate's identity. + /// + /// A continuation's input differs from the paused run's by exactly the + /// reserved ledger keys, so counting either would make every continuation + /// gate read as a new decision and stack a duplicate card for one question — + /// the failure `is_same_gate` exists to prevent. The delivery half of this + /// is pinned beside it; this is the same claim for the #846 key. + #[test] + fn the_outward_ledger_is_not_part_of_a_gates_identity() { + let input = serde_json::json!({ "request": "x" }); + let paused = gate_effect("wf", "publish", &input, "run-1", &[], &[], None); + + let mut continuation = input.clone(); + continuation.as_object_mut().expect("object").insert( + CONTINUATION_PERFORMED_KEY.to_string(), + serde_json::json!([PerformedCall { + node: "notify".into(), + tool: "http_request POST".into(), + result: serde_json::json!({ "status": 201 }), + }]), + ); + let re_reached = gate_effect("wf", "publish", &continuation, "run-2", &[], &[], None); + + assert!( + is_same_gate(&paused, &re_reached), + "a ledger key must not split one decision into two cards" + ); + } + /// A run that delivered nothing writes no reserved key at all, so an /// ordinary continuation's input keeps exactly the shape it always had. #[test] @@ -923,6 +1281,7 @@ mod tests { &serde_json::json!({ "request": "x" }), "run-1", &[delivery("summary", "owner", DeliveryStatus::Sent)], + &[], None, ); // The same gate, re-reached by the continuation the card started: same @@ -933,7 +1292,7 @@ mod tests { .as_object_mut() .expect("object") .remove("approvals"); - let re_reached = gate_effect("digest", "gate", &continuation, "run-2", &[], None); + let re_reached = gate_effect("digest", "gate", &continuation, "run-2", &[], &[], None); assert!( is_same_gate(&paused, &re_reached), @@ -965,7 +1324,7 @@ mod tests { }); let edges = [edge("start", "writer"), edge("writer", "publish")]; - let mut effect = gate_effect("wf", "publish", &Value::Null, "run-1", &[], None); + let mut effect = gate_effect("wf", "publish", &Value::Null, "run-1", &[], &[], None); attach_upstream_content(&mut effect, &output, &edges, "publish"); let content = &effect.payload[PAYLOAD_CONTENT]; @@ -986,7 +1345,7 @@ mod tests { fn a_gate_with_no_upstream_output_gets_an_empty_preview() { let output = serde_json::json!({ "nodes": {} }); let edges = [edge("writer", "publish")]; - let mut effect = gate_effect("wf", "publish", &Value::Null, "run-1", &[], None); + let mut effect = gate_effect("wf", "publish", &Value::Null, "run-1", &[], &[], None); attach_upstream_content(&mut effect, &output, &edges, "publish"); assert_eq!(effect.payload[PAYLOAD_CONTENT], serde_json::json!({})); } @@ -1001,7 +1360,7 @@ mod tests { let edges = [edge("writer", "publish")]; let input = serde_json::json!({ "request": "x" }); - let mut a = gate_effect("wf", "publish", &input, "run-1", &[], None); + let mut a = gate_effect("wf", "publish", &input, "run-1", &[], &[], None); attach_upstream_content( &mut a, &serde_json::json!({ "nodes": { "writer": { "items": ["draft one"] } } }), @@ -1009,7 +1368,7 @@ mod tests { "publish", ); - let mut b = gate_effect("wf", "publish", &input, "run-2", &[], None); + let mut b = gate_effect("wf", "publish", &input, "run-2", &[], &[], None); attach_upstream_content( &mut b, &serde_json::json!({ "nodes": { "writer": { "items": ["a totally different draft"] } } }), @@ -1231,7 +1590,15 @@ mode = "full" input: Value, deliveries: &[DeliveryReport], ) -> ApprovalId { - let effect = gate_effect("gated", "gate", &input, "run-that-paused", deliveries, None); + let effect = gate_effect( + "gated", + "gate", + &input, + "run-that-paused", + deliveries, + &[], + None, + ); let id = rt .approvals .park(rt.id(), effect.clone()) diff --git a/src/workflows/caps/dry_run.rs b/src/workflows/caps/dry_run.rs index ad666e88c..997170e05 100644 --- a/src/workflows/caps/dry_run.rs +++ b/src/workflows/caps/dry_run.rs @@ -100,6 +100,19 @@ impl DryRunTools { #[async_trait] impl ToolInvoker for DryRunTools { async fn invoke(&self, slug: &str, args: Value, _conn: Option<&str>) -> TfResult { + // Issue #846: the replay arm, mirrored from the live invoker. + // + // A dry run cannot reach here through the host's own path — dry runs are + // never continuations, park no gate and stub every effect — so this is + // not load-bearing today. It is here because the alternative is worse + // than redundant: without it, a graph carrying a replay slug would fall + // through to `namespace_of` and fail the node with "not a wired workflow + // tool", which is a dry run reporting a routing failure that the real run + // does not have. The two invokers agreeing about every slug is the + // property a test run's answer is worth anything for. + if let Some(result) = super::super::replay::replayed_result(slug, &args) { + return Ok(result); + } // FAIL-CLOSED grant check FIRST, identical to the live invoker // (`WorkflowToolInvoker::invoke`): a dry run must refuse an ungranted // tool exactly as a real one does, because that refusal is part of the diff --git a/src/workflows/caps/tools.rs b/src/workflows/caps/tools.rs index c14c56cc2..46a9edb96 100644 --- a/src/workflows/caps/tools.rs +++ b/src/workflows/caps/tools.rs @@ -306,6 +306,19 @@ impl ToolInvoker for WorkflowToolInvoker { /// tools are workspace/company scoped, not per-external-account). Threading a /// real connection is a documented follow-on. async fn invoke(&self, slug: &str, args: Value, _conn: Option<&str>) -> TfResult { + // Issue #846: a call this lineage already made, replayed rather than + // repeated. Answered from the arguments the host wrote onto the node at + // translation time; nothing is looked up and nothing executes. + // + // Deliberately ABOVE the grant check, and that is not a hole. The check + // exists to stop a call reaching a capability the company did not grant, + // and this arm reaches no capability at all — there is no tool, no + // namespace and no network. Below the check it would have to be granted + // a namespace of its own, which would be a real widening in exchange for + // nothing. + if let Some(result) = super::super::replay::replayed_result(slug, &args) { + return Ok(result); + } // FAIL-CLOSED grant check FIRST, before any lookup or execution. let Some(namespace) = toolbelt::namespace_of(slug) else { return Err(EngineError::Capability(format!( @@ -504,6 +517,57 @@ mod tests { ); } + /// The replay arm answers a sentinel invocation on an invoker that grants + /// **nothing**, and reaches no capability doing it (issue #846). + /// + /// Both halves matter and neither is provable without the other. Answering + /// on a zero-grant invoker is what proves the arm sits ABOVE the fail-closed + /// grant check — if it sat below, a continuation would have to be granted a + /// namespace for a call it does not make. And the same invoker refusing a + /// real slug in the same test is what proves the arm is a narrow sentinel + /// rather than a hole: nothing else got easier to invoke. + #[tokio::test] + async fn the_replay_sentinel_is_answered_without_a_grant_and_reaches_nothing() { + let dir = tempfile::tempdir().unwrap(); + let audit = tempfile::tempdir().unwrap(); + let security = Arc::new(toolbelt::exec_security( + dir.path(), + crate::harness::policy::PolicyMode::Supervised, + )); + let invoker = WorkflowToolInvoker::new( + security, + dir.path(), + audit.path(), + Vec::new(), + // No grants at all: every real slug is refused fail-closed. + Vec::new(), + &CapabilityFilter::AllowAll, + None, + test_metering(), + ); + + let recorded = json!({ "status": 201, "id": "abc" }); + let encoded = serde_json::to_string(&recorded).unwrap(); + let replayed = invoker + .invoke( + crate::workflows::replay::REPLAY_SLUG, + json!({ crate::workflows::replay::REPLAY_RESULT_KEY: encoded }), + None, + ) + .await + .expect("the sentinel is answered from its own arguments"); + assert_eq!(replayed, recorded); + + // The control: the same invoker still refuses an ungranted real tool. + let refused = invoker + .invoke("shell", json!({ "command": "id" }), None) + .await; + assert!( + matches!(refused, Err(EngineError::Capability(ref m)) if m.contains("not granted")), + "{refused:?}" + ); + } + #[test] fn construction_only_initializes_granted_tool_families() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/workflows/gate.rs b/src/workflows/gate.rs index 15c6ff398..dc6975353 100644 --- a/src/workflows/gate.rs +++ b/src/workflows/gate.rs @@ -169,6 +169,20 @@ pub(crate) struct GatedCall { /// on; the full URL is already in the node's own config for anyone who needs /// it. pub target: Option, + /// The call's authored arguments — the `url` a `web_fetch` will fetch, the + /// recipient a send will reach (issue #846). + /// + /// Carried verbatim and **redacted downstream**, at the same projection that + /// redacts a chat card's payload, rather than filtered here: one denylist, + /// one set of bounds, and no second rule for this surface to drift away from + /// (the discipline `crate::runtime::approval_display` records). + /// + /// Distinct from [`target`](Self::target) rather than derived from it, + /// because they answer different questions and are governed differently: + /// `target` is a one-line destination written to the journal and kept after + /// the decision, which is why it is host-only and never a path or query; + /// this is the call itself, shown so the operator can decide. + pub args: Value, } /// Marks every `tool_call` node whose call the company's [`ApprovalPolicy`] @@ -277,6 +291,9 @@ pub(crate) async fn policy_gates( continue; }; + // Cloned because the card wants the same arguments the policy judged + // (issue #846), and `ToolPolicyRequest` takes them by value. + let card_args = args.clone(); let request = ToolPolicyRequest::new( &slug, args, @@ -305,6 +322,7 @@ pub(crate) async fn policy_gates( slug, reason, target, + args: card_args, }); } @@ -337,7 +355,12 @@ fn call_of(node: &tinyflows::model::Node) -> Option<(String, Value, Option Some(( HTTP_REQUEST_TOOL.to_string(), @@ -412,30 +435,101 @@ fn call_of(node: &tinyflows::model::Node) -> Option<(String, Value, Option Option { - let url = config.get("url").and_then(Value::as_str)?; - let method = config - .get("method") - .and_then(Value::as_str) - .unwrap_or("GET") - .to_uppercase(); - let host = url - .split_once("://") +/// The host a `tool_call` node's arguments name, when they name one +/// (issue #846). +/// +/// **Host only, never the path or query**, on exactly [`http_target`]'s terms +/// and for exactly its reason: this string is written to the durable journal, +/// rendered on the Approvals page and kept after the decision, and a URL's query +/// is a routine place for tokens and signed parameters to sit. The full +/// arguments travel separately on [`GatedCall::args`], where they are redacted +/// by the shared projection before they reach a console. +/// +/// No method, because a `tool_call` has none to state — that is `http_request`'s +/// vocabulary, and borrowing it would put a `GET` on a card for a call that is +/// not an HTTP request. +/// +/// Reads `url` only. Widening this to "any argument that looks like a URL" would +/// mean guessing which of several is *the* destination, and a card that names +/// the wrong one is worse than a card that names none — the operator would +/// authorise against it. +fn tool_target(args: &Value) -> Option { + let url = args.get("url").and_then(Value::as_str)?; + host_of(url) +} + +/// The host component of `url`, or `None` when there is not one to read. +/// +/// Shared by [`tool_target`] and [`http_target`] so the two surfaces cannot +/// disagree about what a host is — in particular about userinfo, which is +/// everything before the **last** `@` and which a host cannot contain. +fn host_of(url: &str) -> Option { + url.split_once("://") .map(|(_, rest)| rest) .and_then(|rest| rest.split(['/', '?', '#']).next()) - // Userinfo is everything before the LAST `@`; a host cannot contain one. .map(|authority| { authority .rsplit_once('@') .map_or(authority, |(_, host)| host) }) - .filter(|host| !host.is_empty()); - match host { + .filter(|host| !host.is_empty()) + .map(str::to_string) +} + +fn http_target(config: &Value) -> Option { + let url = config.get("url").and_then(Value::as_str)?; + let method = config + .get("method") + .and_then(Value::as_str) + .unwrap_or("GET") + .to_uppercase(); + // Userinfo handling lives in `host_of`, shared with `tool_target`. + match host_of(url) { Some(host) => Some(format!("{method} {host}")), None => Some(format!("{method} (destination resolved at run time)")), } } +/// What a paused node's card should say about the call it is stopping — +/// whoever raised the gate (issue #846). +/// +/// # The hole this closes +/// +/// [`GatedCall`] is produced by [`policy_gates`], so it exists only for a node +/// the **company's policy** stopped. A node the **author** stopped with +/// `requires_approval: true` produced no entry, and its card therefore carried +/// no tool, no arguments and no destination — just a node id and the engine's +/// resume payload. On a `full`-tier company, where the policy stops nothing, +/// that is *every* workflow card: the operator is asked to authorise +/// `fetch_bbc` and shown `{"items":[{"json":{}}],"port":null}`. +/// +/// #372 made the same complaint about the chat surface and #375 fixed it there, +/// by carrying the effect's own arguments onto the card. The information was +/// already on the host in that case and it is already on the host in this one — +/// [`call_of`] has read the slug and the args since #460, and only the *reason* +/// was ever policy-specific. So this asks `call_of` the same question for a node +/// nobody's policy stopped, and the card gains everything except the sentence no +/// one wrote. +/// +/// Returns `None` for a node the graph does not contain, or one whose kind makes +/// no classifiable call — an authored gate on a `transform` is a genuine "stop +/// and look at this", with no call to describe, and a card that invented one +/// would be worse than a card that says so. +pub(crate) fn describe_call(graph: &WorkflowGraph, node_id: &str) -> Option { + let node = graph.nodes.iter().find(|node| node.id == node_id)?; + let (slug, args, target) = call_of(node)?; + Some(GatedCall { + node_id: node.id.clone(), + slug, + // Nobody stated one. The console says "the workflow's author asked for a + // person here" in its own words rather than the host inventing a + // policy-shaped sentence for a decision no policy made. + reason: String::new(), + target, + args, + }) +} + /// The synthetic principal a workflow `tool_call` acts as. /// /// Not a roster teammate, and deliberately shaped so it can never collide with @@ -866,4 +960,86 @@ description = "Runs Acme." } } } + + // --- issue #846: an authored gate's card names its call too ------------ + + /// A node the **author** gated is described, not just identified. + /// + /// This is the whole of #846's third defect. `policy_gates` only ever + /// produced a `GatedCall` for a node the company's policy stopped, so on a + /// `full`-tier company — where the policy stops nothing — every workflow + /// card carried a node id and the engine's resume payload and named neither + /// the tool nor the host. #375 fixed exactly this on the chat surface by + /// carrying the call's own arguments; this asks `call_of` the same question + /// for a gate nobody's policy raised. + #[test] + fn an_authored_gate_is_described_from_the_graph() { + let g = graph(vec![Node { + config: json!({ + "slug": "web_fetch", + "args": { "url": "https://www.bbc.com/sport?token=secret" }, + "requires_approval": true, + }), + ..tool_node("fetch_bbc", "web_fetch") + }]); + + let described = describe_call(&g, "fetch_bbc").expect("a tool_call node is describable"); + assert_eq!(described.node_id, "fetch_bbc"); + assert_eq!(described.slug, "web_fetch"); + assert_eq!( + described.args["url"], + "https://www.bbc.com/sport?token=secret" + ); + // Host only. This string is journalled and kept after the decision, so + // it must never carry the query — where a token is a routine thing to + // find. The full arguments travel on `args`, redacted downstream by the + // shared projection. + assert_eq!(described.target.as_deref(), Some("www.bbc.com")); + // Nobody wrote a reason, and the card must not invent a policy-shaped + // one for a decision no policy made. + assert!(described.reason.is_empty()); + } + + /// A gate on a node that calls nothing is described as such. + /// + /// An authored `requires_approval` on a `transform` is a genuine "stop and + /// look at this" with no call behind it, and a card that invented one would + /// be worse than a card that says nothing. + #[test] + fn a_gate_on_a_node_that_calls_nothing_is_not_described() { + let g = graph(vec![Node { + kind: NodeKind::Transform, + ..tool_node("review", "unused") + }]); + assert!(describe_call(&g, "review").is_none()); + assert!(describe_call(&g, "no-such-node").is_none()); + } + + /// A `tool_call` whose URL is still an unresolved template names no host. + /// + /// Node arguments may still be `=`-expressions when this runs — the module + /// docs record that — and a card that printed `=item.json.url` as a + /// destination would be worse than one that prints none. + #[test] + fn an_unresolved_url_yields_no_host() { + let g = graph(vec![Node { + config: json!({ "slug": "web_fetch", "args": { "url": "=item.json.url" } }), + ..tool_node("fetch", "web_fetch") + }]); + let described = describe_call(&g, "fetch").expect("still describable"); + assert_eq!(described.slug, "web_fetch"); + assert!(described.target.is_none(), "{:?}", described.target); + } + + /// Userinfo is not mistaken for a host. + /// + /// `https://user:pw@evil.test/` must name `evil.test`, not `user`. Shared + /// with `http_target` through `host_of` so the two surfaces cannot disagree. + #[test] + fn userinfo_is_not_mistaken_for_the_host() { + assert_eq!( + tool_target(&json!({ "url": "https://user:pw@evil.test/x?q=1" })).as_deref(), + Some("evil.test") + ); + } } diff --git a/src/workflows/mod.rs b/src/workflows/mod.rs index 63496086a..aab67a779 100644 --- a/src/workflows/mod.rs +++ b/src/workflows/mod.rs @@ -36,6 +36,9 @@ mod gated_tool_call_test; /// node reaches the Approvals page and survives the next chat cycle. #[cfg(test)] mod gated_tool_turn_test; +/// Issue #846: a continuation replays the outward calls its lineage already +/// made, instead of making them a second time. +pub mod replay; pub mod runner; pub mod translate; diff --git a/src/workflows/replay.rs b/src/workflows/replay.rs new file mode 100644 index 000000000..76af73327 --- /dev/null +++ b/src/workflows/replay.rs @@ -0,0 +1,701 @@ +//! Issue #846: a continuation must not make an outward call its own lineage +//! already made. +//! +//! # What this is a second half of +//! +//! A paused workflow run is **settled**, not suspended — the engine finishes and +//! reports the gates it stopped at — so approving is a re-run from the trigger +//! with the gate listed in `approvals`. That primitive is deliberate, it is +//! restart-durable for free, and #438 accepted it: the requirement it wrote down +//! was not "stop re-executing" but **"an approval must never cause a message to +//! be sent to a person twice."** +//! +//! #496 met that requirement for the half the host performs with its own hands: +//! an `output` node's report, routed by [`deliver_outputs`](super::delivery) +//! after the engine settles. A ledger rides the approval card, and a reached +//! output node listed in it is skipped rather than dispatched. +//! +//! It does not reach the other half. A `send`, `publish` or `repo_publish` +//! wired as a **`tool_call` node** is performed by the engine, mid-run, through +//! a capability — so there is no post-hoc dispatch for the host to decline. Put +//! such a node upstream of a gate and today the first run sends, the operator +//! approves, and the continuation sends again. Same exposure, same lineage, the +//! other mechanism. +//! +//! # The seam, and why it is this one +//! +//! [`ToolInvoker::invoke`](tinyflows::caps::ToolInvoker) receives +//! `(slug, args, conn)` — no node id, no run state — which +//! [`gate`](super::gate) already records as the reason the *approval* gate could +//! not live there. The same fact rules out a node-keyed skip inside the invoker, +//! and node identity is not negotiable here: keying on `(tool, args)` instead +//! would miss every send whose body an upstream agent node re-generates, which +//! is most of them. +//! +//! What the host does own is the **translation** — it builds the graph per run, +//! which is where [`apply_policy_gates`](super::gate::apply_policy_gates) +//! already writes per-node decisions. So a node the lineage has already called +//! is rewritten *before the run starts* to invoke a host-private sentinel slug +//! carrying its recorded result, and the invoker answers that slug from its +//! arguments without touching the toolbelt. The engine is unchanged, the graph +//! shape is unchanged, and the node still produces output — the same output, +//! because the recorded value is the verbatim capability return and +//! `envelope::wrap` is a pure function of it. +//! +//! # Which nodes, and the line drawn +//! +//! Two kinds, on two different rules, because the two have different amounts of +//! declaration behind them. +//! +//! **`http_request` with a mutating method** — anything but `GET`/`HEAD`. This +//! is the live one. An `http_request` node reaches an arbitrary address today, +//! on every company, and a `POST` upstream of a gate is a second POST on every +//! approval. Idempotency by HTTP method is the standard the method names carry, +//! and it is the only thing the host can read about a call whose destination is +//! authored per node. +//! +//! **`tool_call` whose [`consequence_of`](crate::policy::consequence_of) group +//! is an outward one** — anything but +//! [`EffectGroup::Other`](crate::ports::types::EffectGroup::Other), less +//! [`Reach::Money`](crate::policy::Reach::Money) (below). That is a reader of the +//! one declaration both policy tiers read, not a second table to keep in step +//! with it. +//! +//! ## What that second rule does and does not reach today, stated plainly +//! +//! **Nothing, today.** A workflow `tool_call` can only invoke the four families +//! [`WORKFLOW_TOOL_NAMESPACES`](super::caps) wires — shell, code, web, search — +//! and every slug in them is declared `EffectGroup::Other` except `web_search`, +//! which this rule excludes. So there is no wired slug this arm currently +//! guards, and #846's own example — "put a `send`, `publish` or `repo_publish` +//! node after two gated steps" — describes agent-turn tool families the workflow +//! invoker does not wire at all. +//! +//! That is worth stating rather than quietly implying otherwise, and it is a +//! reason to write the rule now rather than later: it is the guard that has to +//! exist *before* a send-capable namespace is wired into workflows, or wiring +//! one silently re-opens #438 on the day it lands. It costs one arm of one +//! `match`, and it is exercised by this module's tests against the declared +//! table rather than against a wired tool — which is the most that can honestly +//! be claimed for a rule whose subject does not exist yet. +//! +//! ## Two gaps this does NOT close +//! +//! * **`shell`, `curl` and `git_operations`.** All three are `EffectGroup::Other` +//! and all three can reach a counterparty — `curl` to an address, `shell` to +//! anything at all, `git_operations` to a remote. The host cannot tell a +//! `git log` from a `git push`, and replaying every shell node's recorded +//! stdout on every continuation would change the behaviour of the most common +//! effectful node there is on the strength of a guess. Left as it is, +//! deliberately, and filed as issue #850 rather than folded in. +//! * **`web_fetch` and `web_search`.** Read-only, so the three re-executed +//! fetches in #846's own reproduction still re-execute — which is the issue's +//! own reading of them: idempotent reads whose cost is latency and tokens. +//! `web_search` is excluded by the [`Reach::Money`] carve-out rather than by +//! its group, because it is declared `EffectGroup::Spend` for billing and is +//! a read: it reaches nobody and changes nothing, and #438 priced repeated +//! spend as cost rather than as the harm being guarded here. Replaying it +//! would also hand a later run a stale answer it never asked for, which is the +//! argument against replaying a fetch, and the two should not differ. +//! +//! # Two limits, stated rather than hidden +//! +//! * **A truncated result is never replayed.** The recorded value rides the +//! durable approval card, so it is bounded like everything else that does +//! ([`bound_node_output`](crate::ports::bound_node_output)). If bounding would +//! clip it, the node is not recorded and the continuation calls again — a +//! duplicate send is bad, and feeding downstream a silently-clipped receipt as +//! if it were real is worse. The operator is told, via a run notice. +//! * **A per-item fan-out is not replayed.** A `split_out` → `tool_call` node +//! invokes once per item, and the invoker sees no item index, so one recorded +//! result cannot answer N invocations without inventing which. Recording only +//! the single-invocation shape keeps the guard exact where it applies instead +//! of approximate everywhere; the fan-out case behaves as it does today and +//! says so. #496 took the same side of the same trade for a partial delivery +//! fan-out. +//! +//! Both limits are *visible*: [`replay_performed`] returns what it could not +//! guard so the runner can surface it as a notice (issue #638's mechanism), so +//! "this continuation called out again" is something an operator reads rather +//! than something they reconstruct. +//! +//! The complete answer to all of this is checkpointed resume, and it is still +//! not available: tinyflows' `run_resumable` installs a no-op observer and a +//! process-local in-memory checkpointer and takes no cancellation token, so it +//! composes with neither the per-node progress trail (#371) nor the stop signal +//! (#398) this runner is built on, and an approval that arrives after a restart +//! would find no checkpoint. That is engine work in a vendored crate, exactly as +//! #438 recorded. This guard is forward-compatible with it: it is keyed on the +//! node and it withdraws to nothing when the ledger is empty. + +use serde_json::{Value, json}; +use tinyflows::model::{NodeKind, WorkflowGraph}; + +use crate::ports::bound_node_output; + +use crate::runtime::workflow_resume::{PerformedCall, performed_in_input}; + +/// The host-private slug a replayed node invokes instead of its real tool. +/// +/// Namespaced with a `__opencompany` prefix that no toolbelt tool carries and no +/// authoring surface accepts. Even so the invoker's arm is written to be safe +/// against an author who types it anyway: it reaches no capability, executes +/// nothing and returns only what its own arguments carry, so the worst an +/// authored occurrence can do is produce an inert node — strictly less than any +/// grant would already allow. +pub(crate) const REPLAY_SLUG: &str = "__opencompany.already_performed"; + +/// The argument key carrying the recorded result, **JSON-encoded as a string**. +/// +/// Encoded rather than embedded, and this is load-bearing. Node config is walked +/// by [`tinyflows::expr::resolve`] before the node runs, and every leaf string +/// beginning with `=` is evaluated as an expression against the run scope. A +/// recorded result is arbitrary provider data that may contain such a string, so +/// embedding it verbatim would hand a counterparty's response to the expression +/// engine. `serde_json::to_string` of any value yields a document starting with +/// `{`, `[`, `"`, a digit, `t`, `f` or `n` — never `=` — so a single encoded +/// string is inert by construction rather than by escaping rules. +pub(crate) const REPLAY_RESULT_KEY: &str = "result_json"; + +/// The recorded result behind a [`REPLAY_SLUG`] invocation, or `None` for every +/// other slug (issue #846). +/// +/// The whole of what a tool invoker has to know about this module: one +/// comparison and one decode, so both invokers can answer the sentinel +/// identically without either growing a copy of the rules. +/// +/// A sentinel invocation whose argument is missing or is not the JSON the host +/// wrote yields [`Value::Null`] rather than `None`. Falling through to the real +/// toolbelt would be the one outcome this module exists to prevent — the node +/// was rewritten precisely *because* calling out again is the bug — and no live +/// tool answers to this slug anyway, so the fall-through would fail the node +/// rather than send. A null return is a node that produced nothing, which is +/// visible in the run's own output. +pub(crate) fn replayed_result(slug: &str, args: &Value) -> Option { + if slug != REPLAY_SLUG { + return None; + } + let decoded = args + .get(REPLAY_RESULT_KEY) + .and_then(Value::as_str) + .and_then(|encoded| serde_json::from_str(encoded).ok()) + .unwrap_or(Value::Null); + tracing::info!( + recovered = !decoded.is_null(), + "workflow: replaying a call this run's lineage already made; not calling out again \ + (issue #846)" + ); + Some(decoded) +} + +/// A node the continuation could not replay, and why — surfaced to the operator +/// as a run notice rather than left in the host's logs. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct UnreplayableCall { + /// The node that will call out a second time. + pub node_id: String, + /// The tool it will call. + pub slug: String, + /// Operator-facing prose for why the guard did not apply. + pub why: &'static str, +} + +impl UnreplayableCall { + /// The sentence an operator reads, on the run that parked — before they + /// approve, which is the only moment they can still act on it. + pub fn notice(&self) -> String { + format!( + "“{}” ({}) reached outside the company on this run, and approving the gate below it \ + will call it again: {}.", + self.node_id, self.slug, self.why + ) + } +} + +/// Every `tool_call` node in `graph` whose call left the building on this run, +/// paired with the verbatim result it returned (issue #846). +/// +/// Read off the settled run's own `output["nodes"]`, which already holds every +/// completed node's items — so this costs no engine change and no second source +/// of truth. A node that did not complete has no entry and is not recorded; a +/// node the run never reached (everything past the gate) has none either, which +/// is what makes "recorded" mean "actually happened". +/// +/// Returns `(performed, unreplayable)`: what a continuation may replay, and what +/// it may not, so the caller can carry the second half to the operator instead +/// of silently guarding less than it appears to. +pub(crate) fn outward_calls_performed( + graph: &WorkflowGraph, + output: &Value, +) -> (Vec, Vec) { + let nodes = output.get("nodes"); + let mut performed = Vec::new(); + let mut unreplayable = Vec::new(); + + for node in &graph.nodes { + let Some(slug) = outward_call_of(node) else { + continue; + }; + // Not reached, or reached and produced nothing: there is nothing to + // guard and nothing to replay. + let Some(items) = nodes + .and_then(|map| map.get(&node.id)) + .and_then(|node_output| node_output.get("items")) + .and_then(Value::as_array) + .filter(|items| !items.is_empty()) + else { + continue; + }; + + if items.len() > 1 { + unreplayable.push(UnreplayableCall { + node_id: node.id.clone(), + slug, + why: "it runs once per input item, and a single recorded result cannot stand in \ + for several", + }); + continue; + } + // The engine wraps a capability return as `{ json, text, raw }`, so + // `raw` is the verbatim value — replaying it reconstructs this exact + // envelope rather than approximating it. An item without `raw` is not a + // capability envelope and is not something this can faithfully replay. + let Some(raw) = items[0].get("raw") else { + unreplayable.push(UnreplayableCall { + node_id: node.id.clone(), + slug, + why: "its output is not in the engine's capability envelope, so there is no \ + verbatim result to replay", + }); + continue; + }; + + let (bounded, truncated) = bound_node_output(raw); + if truncated { + unreplayable.push(UnreplayableCall { + node_id: node.id.clone(), + slug, + why: "its result is too large to carry on the approval card, and a clipped \ + result must not be replayed as if it were whole", + }); + continue; + } + performed.push(PerformedCall { + node: node.id.clone(), + tool: slug, + result: bounded, + }); + } + + (performed, unreplayable) +} + +/// Rewrites every node this lineage has already called so the continuation +/// replays its recorded result instead of calling again (issue #846). +/// +/// Mutates `graph` in place, beside [`apply_policy_gates`](super::gate::apply_policy_gates) +/// and before compilation, which is the one moment the host holds both the node +/// ids and the trigger input. A first run — anything with no ledger on its input +/// — leaves the graph byte-identical, so every existing run and every existing +/// test is untouched. +/// +/// Returns the node ids it rewrote, for the log line and for the tests that pin +/// this: an empty return is the honest answer for a graph whose ledger names +/// nodes it does not contain (a graph edited between the pause and the +/// approval), where re-calling is the only thing left to do. +pub(crate) fn replay_performed(graph: &mut WorkflowGraph, trigger_input: &Value) -> Vec { + let ledger = performed_in_input(trigger_input); + if ledger.is_empty() { + return Vec::new(); + } + + let mut replayed = Vec::new(); + for node in &mut graph.nodes { + if !matches!(node.kind, NodeKind::ToolCall | NodeKind::HttpRequest) { + continue; + } + let Some(call) = ledger.iter().find(|call| call.node == node.id) else { + continue; + }; + let Value::Object(config) = &mut node.config else { + continue; + }; + let Ok(encoded) = serde_json::to_string(&call.result) else { + // Unserializable is unreachable for a value that arrived by + // deserialization, and re-calling is the safe direction if it ever + // is not: a node that runs twice is the bug being fixed, a node that + // silently returns nothing is a new one. + continue; + }; + + // An `http_request` node becomes a `tool_call` invoking the sentinel. + // + // The kind change is what lets one seam serve both, and it is safe + // because the two node kinds already agree about the only thing that + // leaves the node: every capability node wraps its result in the same + // `{ json, text, raw }` envelope, so a downstream `=item.json.` + // binding reads the same value either way. The alternative was a second + // replay mechanism inside `GuardedHttpClient` — which, like the invoker, + // sees no node id and would have needed its own identity scheme. + node.kind = NodeKind::ToolCall; + // The request descriptor, removed rather than left to rot. Leaving it + // would have the engine resolve a URL and headers for a call that is + // not made, and would leave a reader of the compiled graph unable to + // tell a replayed node from a live one. + for key in ["url", "method", "headers", "body"] { + config.remove(key); + } + + config.insert("slug".to_string(), json!(REPLAY_SLUG)); + config.insert("args".to_string(), json!({ REPLAY_RESULT_KEY: encoded })); + // Nothing to connect to, and leaving a stale ref would have the engine + // resolve a connection for a call that is not made. + config.remove("connection_ref"); + // One invocation, one recorded result. A node left in `per_item` mode + // would replay the same result once per input item; `outward_calls_performed` + // refuses to record a fan-out for exactly that reason, and pinning the + // mode here means a graph edited between the pause and the approval + // cannot re-open the hole. + config.insert("execution".to_string(), json!("once")); + replayed.push(node.id.clone()); + } + + replayed +} + +/// The name of the outward call a node makes — or `None` when the node makes no +/// call, or makes one that reaches nobody outside the company. +/// +/// The name is for the operator and the log line; the *identity* a ledger entry +/// is matched on is always the node id. +/// +/// An `agent` node is deliberately absent, and it is a different question rather +/// than an oversight: its own tool calls park through #395's drain and are +/// decided one at a time, and its re-execution is the token cost #438 already +/// priced and declined to fix here. +fn outward_call_of(node: &tinyflows::model::Node) -> Option { + match node.kind { + NodeKind::ToolCall => { + let slug = node.config.get("slug").and_then(Value::as_str)?; + let args = node + .config + .get("args") + .cloned() + .unwrap_or_else(|| json!({})); + let consequence = crate::policy::consequence_of(slug, &args); + // The residual bucket — "no particular consequence to name on the + // card" — is everything that stays inside the company plus the three + // this module's docs name as an open gap. + if consequence.group.is_unclassified() { + return None; + } + // Billed, but it reaches nobody and changes nothing. See the module + // docs: repeated spend is cost, not the harm being guarded. + if consequence.reach.costs_money() { + return None; + } + Some(slug.to_string()) + } + // Reaches an arbitrary address on every company today, so this is the + // arm that guards a *live* duplicate. Read by method, which is the only + // thing the host knows about a destination the author supplies per node + // — and which is exactly what the method names are for. + NodeKind::HttpRequest => { + let method = node + .config + .get("method") + .and_then(Value::as_str) + .unwrap_or("GET") + .to_uppercase(); + if SAFE_METHODS.contains(&method.as_str()) { + return None; + } + Some(format!("http_request {method}")) + } + _ => None, + } +} + +/// HTTP methods a continuation may repeat: the **safe** ones, not the +/// idempotent ones. +/// +/// `PUT` and `DELETE` are idempotent in the RFC's sense — the server state after +/// N identical requests equals the state after one — and that is deliberately +/// not the property being asked for. A duplicate `DELETE` still fires whatever +/// the endpoint does on receipt, and "the row is still gone" is no comfort if +/// the second request also sent someone a notification. Only `GET` and `HEAD` +/// promise to have done nothing. +const SAFE_METHODS: [&str; 2] = ["GET", "HEAD"]; + +#[cfg(test)] +mod tests { + use super::*; + use crate::ports::run_output::RUN_OUTPUT_MAX_BYTES; + use crate::runtime::workflow_resume::CONTINUATION_PERFORMED_KEY; + use tinyflows::model::Node; + + fn node(id: &str, kind: NodeKind, config: Value) -> Node { + Node { + id: id.to_string(), + kind, + type_version: 1, + name: String::new(), + config, + ports: Vec::new(), + position: None, + } + } + + fn graph(nodes: Vec) -> WorkflowGraph { + WorkflowGraph { + id: Some("wf".to_string()), + nodes, + ..WorkflowGraph::default() + } + } + + /// A settled run's output: one completed node, one capability envelope. + fn settled(node_id: &str, raw: Value) -> Value { + json!({ "nodes": { node_id: { "items": [{ "json": raw, "text": null, "raw": raw }] } } }) + } + + /// A `POST` that already fired is recorded, so a continuation can replay it. + /// + /// The live half of issue #846: an `http_request` node reaches an arbitrary + /// address on every company today, so this is the duplicate that is + /// reachable now rather than the one that becomes reachable when a + /// send-capable namespace is wired. + #[test] + fn a_post_that_already_fired_is_recorded() { + let g = graph(vec![node( + "notify", + NodeKind::HttpRequest, + json!({ "method": "POST", "url": "https://api.test/hooks" }), + )]); + let (performed, unreplayable) = + outward_calls_performed(&g, &settled("notify", json!({ "status": 201 }))); + + assert_eq!(performed.len(), 1, "{performed:?}"); + assert_eq!(performed[0].node, "notify"); + assert_eq!(performed[0].tool, "http_request POST"); + assert_eq!(performed[0].result, json!({ "status": 201 })); + assert!(unreplayable.is_empty(), "{unreplayable:?}"); + } + + /// A `GET` is not recorded, and neither is a read-only tool. + /// + /// The negative control for the classification, and the issue's own reading + /// of its reproduction: the three re-executed `web_fetch` nodes still + /// re-execute, because repeating a read costs latency rather than reaching + /// anybody. `web_search` is the `Reach::Money` carve-out — declared + /// `EffectGroup::Spend` for billing, but still a read. + #[test] + fn reads_are_not_recorded_whatever_they_cost() { + let g = graph(vec![ + node( + "fetch", + NodeKind::HttpRequest, + json!({ "method": "GET", "url": "https://api.test/x" }), + ), + node( + "implicit_get", + NodeKind::HttpRequest, + json!({ "url": "https://api.test/x" }), + ), + node( + "page", + NodeKind::ToolCall, + json!({ "slug": "web_fetch", "args": { "url": "https://www.bbc.com/sport" } }), + ), + node( + "search", + NodeKind::ToolCall, + json!({ "slug": "web_search", "args": { "query": "scores" } }), + ), + ]); + let mut output = json!({ "nodes": {} }); + for id in ["fetch", "implicit_get", "page", "search"] { + output["nodes"][id] = settled(id, json!({ "ok": true }))["nodes"][id].clone(); + } + + let (performed, unreplayable) = outward_calls_performed(&g, &output); + assert!(performed.is_empty(), "{performed:?}"); + assert!(unreplayable.is_empty(), "{unreplayable:?}"); + } + + /// A node the run never reached is not recorded — which is what makes + /// "recorded" mean "actually happened" rather than "is in the graph". + #[test] + fn a_node_the_run_never_reached_is_not_recorded() { + let g = graph(vec![node( + "notify", + NodeKind::HttpRequest, + json!({ "method": "POST", "url": "https://api.test/hooks" }), + )]); + let (performed, unreplayable) = + outward_calls_performed(&g, &json!({ "nodes": { "notify": { "items": [] } } })); + assert!(performed.is_empty(), "{performed:?}"); + assert!(unreplayable.is_empty(), "{unreplayable:?}"); + } + + /// A per-item fan-out is NOT recorded, and says so. + /// + /// The invoker sees no item index, so one recorded result cannot answer N + /// invocations without inventing which. Guarding it approximately would be + /// worse than not guarding it, so this refuses — and surfaces a notice, so + /// the operator learns before they approve rather than afterwards. + #[test] + fn a_fan_out_is_refused_and_surfaced() { + let g = graph(vec![node( + "notify_each", + NodeKind::HttpRequest, + json!({ "method": "POST", "url": "https://api.test/hooks" }), + )]); + let output = json!({ + "nodes": { "notify_each": { "items": [ + { "raw": { "status": 201 } }, + { "raw": { "status": 201 } }, + ] } } + }); + + let (performed, unreplayable) = outward_calls_performed(&g, &output); + assert!(performed.is_empty(), "{performed:?}"); + assert_eq!(unreplayable.len(), 1, "{unreplayable:?}"); + assert_eq!(unreplayable[0].node_id, "notify_each"); + assert!( + unreplayable[0].notice().contains("will call it again"), + "the notice must say what approving does: {}", + unreplayable[0].notice() + ); + } + + /// A result too large for the card is NOT recorded. + /// + /// A duplicate send is bad; feeding a downstream node a silently-clipped + /// receipt as though it were whole is worse, so the guard withdraws rather + /// than degrading — and says so. + #[test] + fn an_oversized_result_is_refused_rather_than_clipped() { + let g = graph(vec![node( + "notify", + NodeKind::HttpRequest, + json!({ "method": "POST", "url": "https://api.test/hooks" }), + )]); + let huge = json!({ "body": "x".repeat(RUN_OUTPUT_MAX_BYTES + 1) }); + let (performed, unreplayable) = outward_calls_performed(&g, &settled("notify", huge)); + + assert!(performed.is_empty(), "{performed:?}"); + assert_eq!(unreplayable.len(), 1, "{unreplayable:?}"); + assert!( + unreplayable[0].why.contains("too large"), + "{}", + unreplayable[0].why + ); + } + + /// The rewrite: a recorded node invokes the sentinel instead of its tool. + #[test] + fn a_recorded_node_is_rewritten_to_replay() { + let mut g = graph(vec![ + node( + "notify", + NodeKind::HttpRequest, + json!({ "method": "POST", "url": "https://api.test/hooks", "body": "hi" }), + ), + node( + "page", + NodeKind::ToolCall, + json!({ "slug": "web_fetch", "args": { "url": "https://www.bbc.com" } }), + ), + ]); + let input = json!({ + CONTINUATION_PERFORMED_KEY: [ + { "node": "notify", "tool": "http_request POST", "result": { "status": 201 } } + ] + }); + + assert_eq!(replay_performed(&mut g, &input), vec!["notify".to_string()]); + + let notify = &g.nodes[0]; + assert_eq!(notify.kind, NodeKind::ToolCall, "the seam is the invoker's"); + assert_eq!(notify.config["slug"], json!(REPLAY_SLUG)); + assert_eq!(notify.config["execution"], json!("once")); + // The request descriptor is gone, so nothing resolves a URL for a call + // that is not made. + for key in ["url", "method", "body"] { + assert!( + notify.config.get(key).is_none(), + "{key} survived the rewrite" + ); + } + // An unrecorded node is untouched — this promotes, it never rewrites + // what it was not told about. + assert_eq!(g.nodes[1].config["slug"], json!("web_fetch")); + } + + /// A first run — no ledger — leaves the graph byte-identical. + /// + /// The claim every existing run and every existing test depends on. + #[test] + fn a_first_run_rewrites_nothing() { + let original = graph(vec![node( + "notify", + NodeKind::HttpRequest, + json!({ "method": "POST", "url": "https://api.test/hooks" }), + )]); + let mut g = original.clone(); + + assert!(replay_performed(&mut g, &json!({ "topic": "q3" })).is_empty()); + assert_eq!( + serde_json::to_value(&g).unwrap(), + serde_json::to_value(&original).unwrap(), + ); + } + + /// The invoker answers the sentinel with the verbatim recorded value. + #[test] + fn the_sentinel_returns_the_recorded_result() { + let encoded = serde_json::to_string(&json!({ "status": 201, "id": "abc" })).unwrap(); + let replayed = replayed_result(REPLAY_SLUG, &json!({ REPLAY_RESULT_KEY: encoded })); + assert_eq!(replayed, Some(json!({ "status": 201, "id": "abc" }))); + + // Every other slug is none of this module's business. + assert_eq!(replayed_result("web_fetch", &json!({ "url": "x" })), None); + // A sentinel with nothing to replay yields null rather than falling + // through to the real toolbelt — falling through is the one outcome + // this exists to prevent. + assert_eq!(replayed_result(REPLAY_SLUG, &json!({})), Some(Value::Null)); + } + + /// A recorded result containing an `=`-prefixed string is replayed + /// **verbatim**, not evaluated as an engine expression. + /// + /// This is why the result is JSON-encoded into a single string rather than + /// embedded in the node config: `tinyflows::expr::resolve` walks a node's + /// config before it runs and evaluates every leaf beginning with `=`, and a + /// recorded result is arbitrary data from a counterparty. The encoding makes + /// that inert by construction — `serde_json::to_string` never yields a + /// document starting with `=`. + #[test] + fn a_recorded_expression_string_is_not_evaluated() { + let hostile = json!({ "note": "=run.trigger.secret" }); + let mut g = graph(vec![node( + "notify", + NodeKind::HttpRequest, + json!({ "method": "POST", "url": "https://api.test/hooks" }), + )]); + let (performed, _) = outward_calls_performed(&g, &settled("notify", hostile.clone())); + let input = json!({ CONTINUATION_PERFORMED_KEY: performed }); + + replay_performed(&mut g, &input); + let args = &g.nodes[0].config["args"]; + + // Nothing in the rewritten config is an `=`-expression: the whole + // recorded value is one JSON string. + let encoded = args[REPLAY_RESULT_KEY] + .as_str() + .expect("encoded as a string"); + assert!(!encoded.starts_with('='), "{encoded}"); + assert_eq!(replayed_result(REPLAY_SLUG, args), Some(hostile)); + } +} diff --git a/src/workflows/runner.rs b/src/workflows/runner.rs index 6e5b4d8f8..b42baeda8 100644 --- a/src/workflows/runner.rs +++ b/src/workflows/runner.rs @@ -223,6 +223,29 @@ async fn run_workflow_inner( } else { super::gate::apply_policy_gates(&mut graph, record, &workflow.id, &ctx.run_id).await }; + // Issue #846: a node whose call already left the building in an earlier run + // of this lineage replays its recorded result instead of calling again. + // Driven entirely off the trigger input's ledger, so a first run rewrites + // nothing and the graph stays byte-identical. + // + // **After the gate pass, and the order is load-bearing.** The gate pass + // classifies a node by its slug; running it second would have it classify + // the host's replay sentinel — an inert slug no policy has an opinion about + // — and either gate a node that does nothing or fail to gate one that does. + // Nothing is lost by this order: a replayed node was necessarily executed by + // an earlier run, so its id is already in the input's `approvals` array and + // the gate it still carries falls straight through. + let replayed = super::replay::replay_performed(&mut graph, &input); + if !replayed.is_empty() { + tracing::info!( + company = %record.id, + workflow = %workflow.id, + run_id = %ctx.run_id, + nodes = ?replayed, + "workflow: this continuation replays calls an earlier run in its lineage already \ + made, rather than repeating them" + ); + } let compiled = tinyflows::compiler::compile(&graph).map_err(map_engine_error)?; // Issue #371: the caller's run id, not a freshly minted one. Correlating the // run's progress events with the `WorkflowRunFinished` the caller journals @@ -653,6 +676,32 @@ async fn run_workflow_inner( // Skipped for a cancelled run, which returns above: an operator who stopped // a run is not asking to be asked about the gates it never reached. (A dry // run also never reaches here — it returned above, having parked nothing.) + // Issue #846: what this run's `tool_call` nodes sent outside the company, + // and what it could not record. Computed here, beside the delivery ledger + // and for the same reason — the card an approval is decided from has to + // carry both, or approving repeats one of them. + // + // Only when the run actually paused: a run that reached the end has no + // continuation coming, so there is nothing to guard against and nothing to + // warn about. + let (performed, unreplayable) = if outcome.pending_approvals.is_empty() { + (Vec::new(), Vec::new()) + } else { + super::replay::outward_calls_performed(&graph, &outcome.output) + }; + for call in &unreplayable { + tracing::warn!( + company = %record.id, + workflow = %workflow.id, + %run_id, + node = %call.node_id, + tool = %call.slug, + why = call.why, + "workflow: an outward call this run made cannot be replayed, so approving a gate \ + below it will repeat it" + ); + notices.push(call.notice()); + } park_pending_gates( delivery.as_ref(), record, @@ -662,7 +711,9 @@ async fn run_workflow_inner( trigger_input: &trigger_input, pending: &outcome.pending_approvals, deliveries: &deliveries, + performed: &performed, gated: &gated, + graph: &graph, // Issue #596: the reached-node output + the graph's edges, so each // parked gate can carry the verbatim upstream content awaiting // sign-off. @@ -760,9 +811,15 @@ struct PausedGates<'a> { pending: &'a [String], /// What this run actually routed (issue #438). deliveries: &'a [crate::ports::DeliveryReport], - /// The policy-raised gates, so a card can say which tool and why. An - /// authored gate has no entry here and its card stays as #395 shipped it. + /// What this run already sent outside the company (issue #846), so + /// approving replays it rather than repeating it. + performed: &'a [crate::runtime::workflow_resume::PerformedCall], + /// The policy-raised gates, so a card can say which tool and **why**. An + /// authored gate has no entry here — nobody stated a reason — and issue #846 + /// reads its call off the graph instead, so the card still names it. gated: &'a [super::gate::GatedCall], + /// The run's graph, for the authored-gate description above (issue #846). + graph: &'a tinyflows::model::WorkflowGraph, /// Issue #596: the run's reached-node output and the graph's edges, so a /// parked gate's card can carry the verbatim upstream content awaiting /// sign-off. Additive to the #460 struct — the pre-existing fields are @@ -817,7 +874,9 @@ async fn park_pending_gates( trigger_input, pending, deliveries, + performed, gated, + graph, // Issue #596: the reached-node output + the graph's edges, so each parked // gate's card can carry the verbatim upstream content awaiting sign-off. output, @@ -843,22 +902,39 @@ async fn park_pending_gates( for node_id in pending { // Issue #460: when the policy is what stopped this node, the card says - // which tool and why. An authored gate carries neither — nobody asked a - // question on its behalf — so it stays exactly as #395 shipped it. - let call = gated - .iter() - .find(|gate| gate.node_id == *node_id) - .map(|gate| crate::runtime::workflow_resume::GateCall { - tool: gate.slug.as_str(), - reason: gate.reason.as_str(), - target: gate.target.as_deref(), - }); + // which tool and why. + // + // Issue #846: when the **author** stopped it, the card still says which + // tool — read off the graph, which has known the node's slug and + // arguments all along. Only the reason is policy-specific, and it is the + // one thing an authored gate genuinely does not have. Falling back rather + // than merging: a policy-raised gate already carries the same call, so + // consulting the graph for it would be a second answer to a question that + // already has one. + let described; + let gate = match gated.iter().find(|gate| gate.node_id == *node_id) { + Some(gate) => Some(gate), + None => { + described = super::gate::describe_call(graph, node_id); + described.as_ref() + } + }; + let call = gate.map(|gate| crate::runtime::workflow_resume::GateCall { + tool: gate.slug.as_str(), + // Empty means "nobody wrote one", which `describe_call` documents; + // the key is then absent from the payload rather than present and + // blank, so a console can tell an unstated reason from an empty one. + reason: Some(gate.reason.as_str()).filter(|reason| !reason.is_empty()), + args: Some(&gate.args), + target: gate.target.as_deref(), + }); let mut effect = crate::runtime::workflow_resume::gate_effect( workflow_id, node_id, trigger_input, run_id, deliveries, + performed, call, ); // Issue #596: enrich the card with the verbatim output of this gate's From 30747c269b77fb5b04d782c511a077b05c3dd60a Mon Sep 17 00:00:00 2001 From: M3gA-Mind Date: Thu, 13 Aug 2026 22:04:11 +0530 Subject: [PATCH 2/2] test(console): type the run fixture against the real wire types MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `typecheck:unit` covers `test/unit`, and the fixture built its delivery rows from an invented shape — `target: null` where the wire says `target?: string`, plus two fields that do not exist. Vitest strips types, so it passed locally and failed the Console lane, which is exactly the blind spot that step exists for. --- frontend/test/unit/workflow-gate-card.test.ts | 26 +++++++------------ 1 file changed, 10 insertions(+), 16 deletions(-) diff --git a/frontend/test/unit/workflow-gate-card.test.ts b/frontend/test/unit/workflow-gate-card.test.ts index 6fc4efdee..72a96d222 100644 --- a/frontend/test/unit/workflow-gate-card.test.ts +++ b/frontend/test/unit/workflow-gate-card.test.ts @@ -3,7 +3,7 @@ import { describe, expect, it } from "vitest"; import { approvalAction, payloadLines } from "@/lib/language"; import { awaitingCount, runTone } from "@/views/workflows/run-health"; import type { ApprovalSummary } from "@/api/types"; -import type { WorkflowRunOutcome } from "@/api/workflows"; +import type { DeliveryReport, WorkflowRunOutcome } from "@/api/workflows"; /** * Issue #846, the two console-side halves. @@ -51,15 +51,18 @@ function run(over: Partial = {}): WorkflowRunOutcome { return { seq: 1, atMillis: 1_000, + workflowId: "daily-sports-news-blog", scheduled: false, - running: false, - cancelled: false, - error: null, deliveries: [], pendingApprovals: [], nodes: [], ...over, - } as WorkflowRunOutcome; + }; +} + +/** A parked report row, in the shape the host actually sends. */ +function parkedDelivery(): DeliveryReport { + return { node: "summary", kind: "owner", status: "pending", detail: "waiting on you" }; } describe("a paused workflow gate says what it is approving (#846)", () => { @@ -142,17 +145,8 @@ describe("a run waiting on a person is not finished (#846)", () => { it("counts parked gates and parked reports together", () => { const both = run({ pendingApprovals: ["gate_a", "gate_b"], - deliveries: [ - { - node: "summary", - kind: "owner", - status: "pending", - detail: "waiting", - target: null, - reason: null, - }, - ], - } as Partial); + deliveries: [parkedDelivery()], + }); expect(awaitingCount(both)).toBe(3); });