Add text-retract stream event for failed attempt recovery

.claude/CLAUDE.md

@@ -245,8 +245,9 @@ put to a person still runs one at a time, and results return in the order asked)
 `maxRoundRetries` (default 2, for a `RetryableJobError`, waiting as the provider's
 retry-after says within 1–60 s), `roundTimeoutMs` (a provider can accept a request and never answer;
 past it the round is abandoned as a retryable failure, so the retries cover it; a round's text and
-reasoning are forwarded only once the attempt settles, since a failed attempt's partial cannot be
-taken back from the accumulated port; reasoning goes out on the turn's own `reasoning` port, a
+reasoning are forwarded live, and an attempt that fails takes back exactly what it forwarded with a
+`text-retract` per port, so the folded ports read as if it never ran and earlier rounds are untouched;
+reasoning goes out on the turn's own `reasoning` port, a
 blank line between rounds, and is absent from the output when no round returned any), and
 budgets — `maxInputTokens`, `maxCostUsd` (refused without a price card,
 since a budget it cannot measure never stops anything), `maxDurationMs` — each ending the turn
@@ -306,7 +307,9 @@ switch. DeepSeek's `replay_reasoning` is a different mechanism: it controls repl
 (they draw on a screen, ask a person, read state only that process holds) but whose model is
 reachable only through a backend binds `AGENT_ROUND_RUNNER` on the run's registry: every round is
 then handed to it as a `ToolCallingTaskInput` and answered as a `ToolCallingTaskOutput`, while the
-loop and every tool stay in-process. A serialized graph cannot carry the closures, so without it a
+loop and every tool stay in-process. The runner streams back through its `AgentRoundContext`:
+text, reasoning, and `onTextRetract` for text its own call withdrew, which the turn caps at what
+that attempt forwarded. A serialized graph cannot carry the closures, so without it a
 host chooses between tools that work and a model it can reach. A registry binding, not an input,
 for the same reason as `AGENT_APPROVAL_OPT_OUT`: where a turn's model calls go is the host's call.
 
@@ -318,7 +321,7 @@ task: `normalizeHistoryForModel` / `trimHistoryForModel` (`ChatHistory.ts`), and
 unit-tested in `packages/ai/src/task/__tests__/`: `AgentTurnBudget` (tokens, cost, deadline —
 `exhausted()` answers the answer path, `mustStopLoop()` the loop, so changing what an answered
 turn reports cannot move where the loop stops), `AgentRoundAttempt` (one round's retries,
-timeout and held deltas), `AgentTurnPolicy` (submit tool, reminders, review, time-left notice)
+timeout, and the live deltas a failed attempt takes back), `AgentTurnPolicy` (submit tool, reminders, review, time-left notice)
 and `AgentToolRound` (dispatching a round's calls and building their results).
 
 `validateModelAuthoredSchema` (`ModelAuthoredSchema.ts`) bounds a JSON Schema a model wrote

docs/technical/04-dataflow-and-streaming.md

@@ -348,12 +348,16 @@ All streaming data flows through the `StreamEvent` discriminated union type:
 ```typescript
 type StreamEvent<Output = Record<string, any>> =
   | StreamTextDelta
+  | StreamTextRetract
   | StreamObjectDelta
   | StreamSnapshot<Output>
   | StreamFinish<Output>
   | StreamError;
 ```
 
+(Abridged: the union also carries `binary-delta`, `refusal`, `usage`, `phase` and `tool-call`
+events; see `StreamTypes.ts`.)
+
 ### StreamTextDelta
 
 ```typescript
@@ -366,6 +370,29 @@ interface StreamTextDelta {
 
 Used with `x-stream: "append"`. Each event carries a fragment of text that should be appended to the accumulated result for the named port.
 
+### StreamTextRetract
+
+```typescript
+interface StreamTextRetract {
+  type: "text-retract";
+  port: string; // Output port name
+  length: number; // UTF-16 code units to take back
+}
+```
+
+Append mode only. Removes the last `length` characters streamed on `port` in this run, clamped at
+the start; later `text-delta`s append after what remains. Every consumer folding a port's text
+truncates with `retractText(acc, length)`, which removes `retractionLength(length)` characters (a
+fraction rounds down, and anything not positive removes nothing). A
+producer that streams text it may yet abandon uses it: `AgentTask` forwards each round live and,
+when an attempt fails and is retried, takes back exactly what that attempt sent, so the folded port
+reads as if the attempt never ran. Dataflow edges carrying the port carry the retraction too.
+
+A port whose text was already teed to a cache sink cannot be truncated there: the processor fails
+that port's sink (discarding the partial write) and settles the port inline from the accumulator
+for the rest of the run. With no accumulator either, the run fails at `finish` unless the finish
+payload states the port's value itself.
+
 ### StreamObjectDelta
 
 ```typescript
@@ -451,6 +478,11 @@ text-delta:
   - Emit "stream_chunk" on the task
   - Update progress (asymptotic curve: 1 - e^(-0.05*chunkCount))
 
+text-retract:
+  - Truncate the port's accumulated text (if shouldAccumulate)
+  - Fail that port's cache sink, if one was teed (see StreamTextRetract)
+  - Emit "stream_chunk" on the task
+
 object-delta:
   - Accumulate per-port (if shouldAccumulate)
   - Update runOutputData progressively

examples/cli/src/agent/runAgentChat.ts

@@ -7,6 +7,7 @@
 import type { AgentApprovalMode, AgentTaskOutput, ChatMessage, ToolDefinition } from "@workglow/ai";
 import { AGENT_APPROVAL_OPT_OUT, AgentTask } from "@workglow/ai";
 import type { StreamEvent } from "@workglow/task-graph";
+import { retractionLength } from "@workglow/task-graph";
 import {
   globalServiceRegistry,
   HUMAN_CONNECTOR,
@@ -202,14 +203,26 @@ async function runTurn(
     // here it would read as part of the answer.
     if (event.type === "text-delta" && (event.port ?? "text") === "text") {
       transcript.delta(event.textDelta);
+    } else if (
+      event.type === "text-retract" &&
+      (event.port ?? "text") === "text" &&
+      retractionLength(event.length) > 0 &&
+      !controller.signal.aborted
+    ) {
+      // An attempt that failed partway takes its text back. What it printed is
+      // already in the scrollback and cannot be erased, so say it no longer
+      // counts; the retry's reply follows on a line of its own. An interrupted
+      // turn retracts too, but it did not fail: the "interrupted" note says so.
+      transcript.note("  · that attempt failed; the partial reply above is discarded");
     }
   });
   const offProgress = task.subscribe("progress", (_progress, message) => {
-    // Only the tool lines are worth a row: "Thinking" is what the blank space
-    // between the prompt and the first token already says.
+    // Only the tool and retry lines are worth a row: "Thinking" is what the
+    // blank space between the prompt and the first token already says.
     if (!message || message === phase) return;
     phase = message;
     if (message.startsWith("Running ")) transcript.note(`  · ${message.slice("Running ".length)}`);
+    else if (message.startsWith("Retrying ")) transcript.note(`  · ${message.toLowerCase()}`);
   });
 
   try {

examples/cli/src/run-events/RunEventTypes.ts

@@ -61,6 +61,16 @@ export type RunEvent =
   | { readonly k: "text"; readonly id: string; readonly delta: string }
   /** A model's thinking, streamed beside its answer; never folded into `text`. */
   | { readonly k: "reasoning"; readonly id: string; readonly delta: string }
+  /**
+   * Takes back the last `length` characters of a row's `text` or `reasoning`:
+   * a model call that streamed and then failed, retracted before its retry.
+   */
+  | {
+      readonly k: "retract";
+      readonly id: string;
+      readonly port: "text" | "reasoning";
+      readonly length: number;
+    }
   | { readonly k: "messages"; readonly id: string; readonly messages: unknown }
   | {
       readonly k: "human_request";

examples/cli/src/run-events/projectRunEvents.test.ts

@@ -4,7 +4,8 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-import { Task, Workflow, type IExecuteContext } from "@workglow/task-graph";
+import type { IExecuteContext, StreamEvent } from "@workglow/task-graph";
+import { Task, Workflow } from "@workglow/task-graph";
 import type { DataPortSchema } from "@workglow/util/schema";
 import { describe, expect, it } from "vitest";
 import type { RunEvent } from "./RunEventTypes";
@@ -94,7 +95,41 @@ class ThinkingTask extends Task<Record<string, never>, { text: string; reasoning
   }
 }
 
+/** Streams a partial, takes it back, then streams the reply, as a retried `AgentTask` round does. */
+class RetractingTask extends Task<Record<string, never>, { text: string; reasoning: string }> {
+  static override readonly type = "RetractingTask";
+  static override readonly title = "Retracting task";
+  static override readonly category = "Test";
+  static override readonly cacheable = false;
+  static override inputSchema(): DataPortSchema {
+    return EMPTY;
+  }
+  static override outputSchema(): DataPortSchema {
+    return STREAMED_PORTS;
+  }
+  async *executeStream(): AsyncIterable<StreamEvent<{ text: string; reasoning: string }>> {
+    yield { type: "text-delta" as const, port: "text", textDelta: "Oops" };
+    yield { type: "text-retract" as const, port: "text", length: 4 };
+    yield { type: "text-delta" as const, port: "text", textDelta: "Hello." };
+    yield { type: "finish" as const, data: {} as { text: string; reasoning: string } };
+  }
+}
+
 describe("projectRunEvents streamed text", () => {
+  it("reports a retraction as its own kind, naming the port", async () => {
+    const { events, sink } = collector();
+    const workflow = new Workflow();
+    workflow.pipe(new RetractingTask() as never);
+    const stop = projectRunEvents(workflow.graph, sink);
+    await workflow.run({});
+    stop();
+
+    const text = events.flatMap((e) =>
+      e.k === "text" ? [e.delta] : e.k === "retract" ? [`-${e.port}:${e.length}`] : []
+    );
+    expect(text).toEqual(["Oops", "-text:4", "Hello."]);
+  });
+
   it("reports a task's text and reasoning as separate kinds", async () => {
     const { events, sink } = collector();
     const workflow = new Workflow();

examples/cli/src/run-events/projectRunEvents.ts

@@ -5,7 +5,7 @@
  */
 
 import type { ITask, StreamEvent, TaskGraph, TaskIdType, Usage } from "@workglow/task-graph";
-import { classifyTextDelta } from "../ui/model/streamedText";
+import { classifyTextDelta, classifyTextRetract } from "../ui/model/streamedText";
 import { cliTaskLabel } from "../ui/taskGraphCliSubscriptions";
 import { projectTaskSubtree } from "./projectSubtrees";
 import type { RunEventSink } from "./runEventChannel";
@@ -130,6 +130,16 @@ export function projectRunEvents(
           delta: classified.delta,
         });
       }
+    } else if (event.type === "text-retract") {
+      const classified = classifyTextRetract(event);
+      if (classified) {
+        sink.emit({
+          k: "retract",
+          id: String(taskId),
+          port: classified.port,
+          length: classified.length,
+        });
+      }
     } else if (event.type === "object-delta" && (event as { port?: string }).port === "messages") {
       const task = graph.getTask(taskId);
       sink.emit({

examples/cli/src/ui/model/streamedText.test.ts

@@ -48,6 +48,42 @@ describe("foldTextDelta", () => {
   });
 });
 
+describe("foldTextDelta on a text-retract", () => {
+  it("takes back the tail of its own port, so the retry appends after what remains", () => {
+    const out = fold([
+      { type: "text-delta", port: "reasoning", textDelta: "Thinking" },
+      { type: "text-delta", port: "text", textDelta: "First. " },
+      { type: "text-delta", port: "text", textDelta: "Oops" },
+      { type: "text-retract", port: "text", length: 4 },
+      { type: "text-delta", port: "text", textDelta: "Second." },
+    ]);
+    expect(out).toEqual({ text: "First. Second.", reasoning: "Thinking" });
+  });
+
+  it("retracts reasoning on its own port, and clamps at the start", () => {
+    const out = fold([
+      { type: "text-delta", port: "reasoning", textDelta: "abc" },
+      { type: "text-delta", port: "text", textDelta: "kept" },
+      { type: "text-retract", port: "reasoning", length: 10 },
+    ]);
+    expect(out).toEqual({ text: "kept", reasoning: "" });
+  });
+
+  it("returns the same object when there is nothing to take back", () => {
+    const prev: StreamedText = { text: "a", reasoning: "" };
+    expect(foldTextDelta(prev, { type: "text-retract", port: "text", length: 0 })).toBe(prev);
+    expect(foldTextDelta(prev, { type: "text-retract", port: "reasoning", length: 3 })).toBe(prev);
+  });
+
+  it("removes the whole number of characters retractText removes", () => {
+    const prev: StreamedText = { text: "Pabc", reasoning: "" };
+    expect(foldTextDelta(prev, { type: "text-retract", port: "text", length: 1.5 }).text).toBe(
+      "Pab"
+    );
+    expect(foldTextDelta(prev, { type: "text-retract", port: "text", length: 0.5 })).toBe(prev);
+  });
+});
+
 describe("visibleReasoning", () => {
   it("shows thinking until the answer starts", () => {
     expect(visibleReasoning({ text: "", reasoning: "hmm" })).toBe("hmm");

examples/cli/src/ui/model/streamedText.ts

@@ -24,6 +24,8 @@ export interface TextDeltaLike {
   readonly type: string;
   readonly port?: string;
   readonly textDelta?: string;
+  /** On a `text-retract`: how many trailing characters to take back. */
+  readonly length?: number;
 }
 
 /** Lines of a streaming panel shown at once; the oldest scroll off the top. */
@@ -48,11 +50,46 @@ export function classifyTextDelta(event: TextDeltaLike): ClassifiedDelta | undef
   return { port: event.port === "reasoning" ? "reasoning" : "text", delta };
 }
 
+/** A text retraction on a port worth showing: which one, and how much it takes back. */
+export interface ClassifiedRetract {
+  readonly port: "text" | "reasoning";
+  readonly length: number;
+}
+
+/**
+ * Reads a stream event as a `text-retract`, or `undefined` when it is not one
+ * that takes anything back. Ports are read as {@link classifyTextDelta} reads
+ * them, so a retraction lands on the text its deltas went to.
+ */
+export function classifyTextRetract(event: TextDeltaLike): ClassifiedRetract | undefined {
+  if (event.type !== "text-retract") return undefined;
+  // Rounded as `retractText` rounds it, so every fold downstream of this one,
+  // the web console's included, removes the same whole number of characters.
+  const length = Math.floor(event.length ?? 0);
+  if (!(length > 0)) return undefined;
+  return { port: event.port === "reasoning" ? "reasoning" : "text", length };
+}
+
+/** `text` without its last `length` characters, clamped at the start. */
+function dropTail(text: string, length: number): string {
+  return text.slice(0, Math.max(0, text.length - length));
+}
+
 /**
- * Folds one stream event into the accumulated text. Returns `prev` itself when
- * nothing changed so a state setter can bail out of a re-render.
+ * Folds one stream event into the accumulated text: a `text-delta` appends and
+ * a `text-retract` takes its tail back. Returns `prev` itself when nothing
+ * changed so a state setter can bail out of a re-render.
  */
 export function foldTextDelta(prev: StreamedText, event: TextDeltaLike): StreamedText {
+  const retract = classifyTextRetract(event);
+  if (retract) {
+    if (retract.port === "text") {
+      if (prev.text.length === 0) return prev;
+      return { text: dropTail(prev.text, retract.length), reasoning: prev.reasoning };
+    }
+    if (prev.reasoning.length === 0) return prev;
+    return { text: prev.text, reasoning: dropTail(prev.reasoning, retract.length) };
+  }
   const classified = classifyTextDelta(event);
   if (!classified) return prev;
   return classified.port === "text"

examples/cli/src/ui/rows/ChatTaskRow.tsx

@@ -83,15 +83,16 @@ export function ChatTaskRow({ task, line }: TaskRowProps): React.ReactElement {
     () => messagesFromTask(task)
   );
   // `streamed` is the in-progress assistant text (and thinking, kept apart by
-  // port) for the *current* turn. Appends on every text-delta; resets whenever
+  // port) for the *current* turn. Appends on every text-delta, drops the tail a
+  // text-retract takes back (a failed attempt's partial); resets whenever
   // history grows (meaning the previous turn's assistant message just landed,
   // or a new user turn has started), and on stream_end.
   const [streamed, setStreamed] = useState<StreamedText>(EMPTY_STREAMED_TEXT);
   const usageLine = useTaskUsageLine(task);
 
   useEffect(() => {
     const onChunk = (event: StreamEvent): void => {
-      if (event.type === "text-delta") {
+      if (event.type === "text-delta" || event.type === "text-retract") {
         setStreamed((prev) => foldTextDelta(prev, event));
       } else if (event.type === "object-delta" && event.port === "messages") {
         // Each object-delta marks a turn boundary — clear the per-turn

examples/cli/src/web/client/state.test.ts

@@ -72,6 +72,19 @@ describe("reduceRunEvent", () => {
     expect(state.rows.get("t1")!.streamText).toBe("Hello");
   });
 
+  it("takes back the tail a retraction names, on its own port", () => {
+    const state = apply([
+      added("t1"),
+      { k: "reasoning", id: "t1", delta: "Hmm" },
+      { k: "text", id: "t1", delta: "First. Oops" },
+      { k: "retract", id: "t1", port: "text", length: 4 },
+      { k: "text", id: "t1", delta: "Second." },
+      { k: "retract", id: "t1", port: "reasoning", length: 99 },
+    ]);
+    expect(state.rows.get("t1")!.streamText).toBe("First. Second.");
+    expect(state.rows.get("t1")!.reasoningText).toBe("");
+  });
+
   it("drops a row the run released", () => {
     const state = apply([added("t1"), { k: "task_removed", id: "t1" }]);
     expect(state.rows.has("t1")).toBe(false);

examples/cli/src/web/client/state.ts

@@ -229,6 +229,18 @@ export function reduceRunEvent(
       if (!row) return state;
       return withRow(state, event.id, { reasoningText: row.reasoningText + event.delta });
     }
+    case "retract": {
+      const row = state.rows.get(event.id);
+      if (!row) return state;
+      const held = event.port === "text" ? row.streamText : row.reasoningText;
+      const kept = held.slice(0, Math.max(0, held.length - event.length));
+      if (kept === held) return state;
+      return withRow(
+        state,
+        event.id,
+        event.port === "text" ? { streamText: kept } : { reasoningText: kept }
+      );
+    }
     case "messages":
       return withRow(state, event.id, { messages: event.messages });
     case "usage":