History

Address macroscope + cursor follow-up review findings

- ClaudeAdapter: dedupe workflow agents/phases by index (last write wins) before applying entry caps, so repeated slot updates cannot exhaust the cap and freeze later agents stale; restrict sessionUrl to http(s) so a hostile tool result cannot smuggle a javascript: href - workflowUi: guard sessionUrl scheme again at both anchor render sites (defense in depth for payloads persisted before the adapter filter) - Ingestion: JSON-encode the (threadId, taskId) tuple in stable workflow activity ids to remove delimiter-ambiguity collisions - workflow-logic: order same-timestamp activities by provider sequence then lifecycle rank so task.completed can never apply before its task.started - WorkflowPanel/Card: scope phase and agent row keys by taskId so switching runs remounts rows instead of leaking expanded transcript state and cursors across runs - workflowUi: phase progress counter counts settled (done + error) agents instead of reporting 0/1 for an errored terminal phase - client-runtime: readJournal query polls every 4s while mounted so the Logs tab picks up new results during a live run - mobile: keep workflow task.completed rows (only per-tick progress is suppressed) — with no workflow card on mobile it is the only signal a workflow finished or failed

Theo Browne committed 4 months ago

Commit 0eaeb941 parent 1610d399 files · +122 −30Browse files

Unified · Split

9 files
  1. apps/mobile/src/lib/threadActivity.ts +4 −2
  2. apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +10 −5
  3. apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +4 −2
  4. apps/server/src/provider/Layers/ClaudeAdapter.ts +30 −9
  5. apps/web/src/components/workflow/WorkflowPanel.tsx +9 −4
  6. apps/web/src/components/workflow/WorkflowRunCard.tsx +10 −4
  7. apps/web/src/components/workflow/workflowUi.tsx +20 −1
  8. apps/web/src/workflow-logic.ts +31 −3
  9. packages/client-runtime/src/state/workflow.ts +4 −0
apps/mobile/src/lib/threadActivity.ts+4−2
@@ -281,11 +281,13 @@ function deriveWorkLogEntries( · 280 unmodified lines
281281 if (activity.kind === "task.started") continue;
282282 // Workflow snapshot/meta activities back the desktop workflow card; on
283283 // mobile they would render as ever-mutating raw rows, so skip them —
284− // along with the per-tick task rows the workflow owns.
284+ // along with the per-tick progress rows the workflow owns. task.completed
285+ // stays: with no workflow card here it is mobile's only signal that a
286+ // workflow finished, failed, or was stopped.
285287 if (activity.kind === "task.workflow-updated") continue;
286288 if (activity.kind === "task.workflow-meta") continue;
287289 if (
288− (activity.kind === "task.progress" || activity.kind === "task.completed") &&
290+ activity.kind === "task.progress" &&
289291 workflowTaskIds.size > 0 &&
290292 activityBelongsToWorkflow(activity, workflowTaskIds)
291293 ) {
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts+10−5
@@ -3122,7 +3122,8 @@ describe("ProviderRuntimeIngestion", () => { · 3121 unmodified lines
31223122
31233123 const thread = await waitForThread(harness.readModel, (entry) =>
31243124 entry.activities.some(
3125− (activity: ProviderRuntimeTestActivity) => activity.id === "workflow:thread-1:task-wf-1",
3125+ (activity: ProviderRuntimeTestActivity) =>
3126+ activity.id === `workflow:${JSON.stringify(["thread-1", "task-wf-1"])}`,
31263127 ),
31273128 );
31283129
@@ -3132,7 +3133,8 @@ describe("ProviderRuntimeIngestion", () => { · 3 unmodified lines
31323133 expect(perTick?.kind).toBe("task.progress");
31333134
31343135 const snapshot = thread.activities.find(
3135− (activity: ProviderRuntimeTestActivity) => activity.id === "workflow:thread-1:task-wf-1",
3136+ (activity: ProviderRuntimeTestActivity) =>
3137+ activity.id === `workflow:${JSON.stringify(["thread-1", "task-wf-1"])}`,
31363138 );
31373139 expect(snapshot?.kind).toBe("task.workflow-updated");
31383140 const payload =
@@ -3165,11 +3167,13 @@ describe("ProviderRuntimeIngestion", () => { · 26 unmodified lines
31653167
31663168 const thread = await waitForThread(harness.readModel, (entry) =>
31673169 entry.activities.some(
3168− (activity: ProviderRuntimeTestActivity) => activity.id === "workflow-meta:thread-1:task-wf-meta",
3170+ (activity: ProviderRuntimeTestActivity) =>
3171+ activity.id === `workflow-meta:${JSON.stringify(["thread-1", "task-wf-meta"])}`,
31693172 ),
31703173 );
31713174 const meta = thread.activities.find(
3172− (activity: ProviderRuntimeTestActivity) => activity.id === "workflow-meta:thread-1:task-wf-meta",
3175+ (activity: ProviderRuntimeTestActivity) =>
3176+ activity.id === `workflow-meta:${JSON.stringify(["thread-1", "task-wf-meta"])}`,
31733177 );
31743178 expect(meta?.kind).toBe("task.workflow-meta");
31753179 const payload =
@@ -3226,7 +3230,8 @@ describe("ProviderRuntimeIngestion", () => { · 50 unmodified lines
32263230 const drainedThread = drained.threads.find((entry) => entry.id === ThreadId.make("thread-1"));
32273231
32283232 const snapshots = drainedThread?.activities.filter(
3229− (activity: ProviderRuntimeTestActivity) => activity.id === "workflow:thread-1:task-wf-2",
3233+ (activity: ProviderRuntimeTestActivity) =>
3234+ activity.id === `workflow:${JSON.stringify(["thread-1", "task-wf-2"])}`,
32303235 );
32313236 expect(snapshots).toHaveLength(1);
32323237 const payload =
apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts+4−2
@@ -167,12 +167,14 @@ function maxCheckpointTurnCount( · 166 unmodified lines
167167 * Stable per-task activity ids: workflow snapshot/meta activities are
168168 * upserted (one projection row per run), not appended per progress tick.
169169 */
170+// JSON-encode the (threadId, taskId) tuple: both are free-form strings, so a
171+// bare `:`-joined key would let distinct pairs collide on the upsert id.
170172 function workflowActivityId(threadId: ThreadId, taskId: string): EventId {
171− return EventId.make(`workflow:${threadId}:${taskId}`);
173+ return EventId.make(`workflow:${JSON.stringify([threadId, taskId])}`);
172174 }
173175
174176 function workflowMetaActivityId(threadId: ThreadId, taskId: string): EventId {
175− return EventId.make(`workflow-meta:${threadId}:${taskId}`);
177+ return EventId.make(`workflow-meta:${JSON.stringify([threadId, taskId])}`);
176178 }
177179
178180 function truncateDetail(value: string, limit = 180): string {
apps/server/src/provider/Layers/ClaudeAdapter.ts+30−9
@@ -831,8 +831,11 @@ function normalizeWorkflowProgress( · 830 unmodified lines
831831 if (!Array.isArray(value) || value.length === 0) {
832832 return undefined;
833833 }
834− const agents: Array<WorkflowAgentProgressEntry> = [];
835− const phases: Array<WorkflowPhaseProgressEntry> = [];
834+ // Last write wins per index: the snapshot may re-emit an agent/phase slot
835+ // (retries, later progress), and the cap must count unique slots, not raw
836+ // entries, or repeats would freeze later agents in a stale state.
837+ const agentsByIndex = new Map<number, WorkflowAgentProgressEntry>();
838+ const phasesByIndex = new Map<number, WorkflowPhaseProgressEntry>();
836839 const logs: Array<WorkflowLogProgressEntry> = [];
837840 for (const raw of value) {
838841 if (raw === null || typeof raw !== "object" || Array.isArray(raw)) {
@@ -842,8 +845,11 @@ function normalizeWorkflowProgress( · 3 unmodified lines
842845 switch (entry.type) {
843846 case "workflow_agent": {
844847 const agent = normalizeWorkflowAgentEntry(entry);
845− if (agent && agents.length < MAX_WORKFLOW_AGENT_ENTRIES) {
846− agents.push(agent);
848+ if (
849+ agent &&
850+ (agentsByIndex.has(agent.index) || agentsByIndex.size < MAX_WORKFLOW_AGENT_ENTRIES)
851+ ) {
852+ agentsByIndex.set(agent.index, agent);
847853 }
848854 break;
849855 }
@@ -853,9 +859,9 @@ function normalizeWorkflowProgress( · 3 unmodified lines
853859 if (
854860 index !== undefined &&
855861 title !== undefined &&
856− phases.length < MAX_WORKFLOW_PHASE_ENTRIES
862+ (phasesByIndex.has(index) || phasesByIndex.size < MAX_WORKFLOW_PHASE_ENTRIES)
857863 ) {
858− phases.push({
864+ phasesByIndex.set(index, {
859865 type: "workflow_phase",
860866 index,
861867 title,
@@ -880,7 +886,7 @@ function normalizeWorkflowProgress( · 18 unmodified lines
880886 // Narration is append-only upstream; keep the newest lines when clipping.
881887 const clippedLogs =
882888 logs.length > MAX_WORKFLOW_LOG_ENTRIES ? logs.slice(-MAX_WORKFLOW_LOG_ENTRIES) : logs;
883− const entries = [...phases, ...agents, ...clippedLogs];
889+ const entries = [...phasesByIndex.values(), ...agentsByIndex.values(), ...clippedLogs];
884890 return entries.length > 0 ? entries : undefined;
885891 }
886892
@@ -893,6 +899,21 @@ function readClaudeWorkflowProgress( · 6 unmodified lines
893899 return normalizeWorkflowProgress(raw);
894900 }
895901
902+function workflowHttpUrl(value: unknown): string | undefined {
903+ const text = workflowString(value);
904+ if (text === undefined) {
905+ return undefined;
906+ }
907+ // Clients render this into an anchor href — restrict to web URLs so a
908+ // hostile tool result cannot smuggle a javascript:/file: scheme through.
909+ try {
910+ const parsed = new URL(text);
911+ return parsed.protocol === "https:" || parsed.protocol === "http:" ? text : undefined;
912+ } catch {
913+ return undefined;
914+ }
915+}
916+
896917 function normalizeWorkflowRunHandles(
897918 toolUseResult: Record<string, unknown>,
898919 ): WorkflowRunHandles | undefined {
@@ -917,8 +938,8 @@ function normalizeWorkflowRunHandles( · 18 unmodified lines
917938 ...(workflowString(toolUseResult.transcriptDir) !== undefined
918939 ? { transcriptDir: workflowString(toolUseResult.transcriptDir) }
919940 : {}),
920− ...(workflowString(toolUseResult.sessionUrl) !== undefined
921− ? { sessionUrl: workflowString(toolUseResult.sessionUrl) }
941+ ...(workflowHttpUrl(toolUseResult.sessionUrl) !== undefined
942+ ? { sessionUrl: workflowHttpUrl(toolUseResult.sessionUrl) }
922943 : {}),
923944 ...(workflowString(toolUseResult.warning) !== undefined
924945 ? { warning: workflowString(toolUseResult.warning) }
apps/web/src/components/workflow/WorkflowPanel.tsx+9−4
@@ -30,7 +30,12 @@ import { · 29 unmodified lines
3030 workflowRunTitle,
3131 } from "~/workflow-logic";
3232 import { Button } from "../ui/button";
33−import { AgentRowContent, PhaseHeader, WorkflowStatusChip } from "./workflowUi";
33+import {
34+ AgentRowContent,
35+ PhaseHeader,
36+ WorkflowStatusChip,
37+ safeWorkflowSessionUrl,
38+} from "./workflowUi";
3439 import { useCopyToClipboard } from "../../hooks/useCopyToClipboard";
3540
3641 type WorkflowTabId = "run" | "script" | "logs";
@@ -199,7 +204,7 @@ function RunTab({ · 162 unmodified lines
199204 remote: boolean;
200205 }): ReactElement {
201206 if (remote) {
202− const sessionUrl = run.handles?.sessionUrl;
207+ const sessionUrl = safeWorkflowSessionUrl(run.handles?.sessionUrl);
203208 if (sessionUrl === undefined) {
204209 return <MutedBody text="Running in the cloud" />;
205210 }
@@ -223,11 +228,11 @@ function RunTab({ · 17 unmodified lines
223228 return (
224229 <div className="space-y-px">
225230 {run.phases.map((phase) => (
226− <div key={phase.index}>
231+ <div key={`${run.taskId}:${phase.index}`}>
227232 <PhaseHeader phase={phase} />
228233 {phase.agents.map((agent) => (
229234 <ExpandableAgentRow
230− key={agent.index}
235+ key={`${run.taskId}:${agent.index}`}
231236 agent={agent}
232237 environmentId={environmentId}
233238 transcriptDir={transcriptDir}
apps/web/src/components/workflow/WorkflowRunCard.tsx+10−4
@@ -11,7 +11,13 @@ import { · 10 unmodified lines
1111 workflowRunTitle,
1212 } from "~/workflow-logic";
1313 import { Button } from "../ui/button";
14−import { AgentRowContent, PhaseHeader, WorkflowStatusChip, agentRollupLabel } from "./workflowUi";
14+import {
15+ AgentRowContent,
16+ PhaseHeader,
17+ WorkflowStatusChip,
18+ agentRollupLabel,
19+ safeWorkflowSessionUrl,
20+} from "./workflowUi";
1521
1622 const MAX_CARD_AGENT_ROWS = 8;
1723
@@ -89,10 +95,10 @@ export function WorkflowRunCard(props: { · 71 unmodified lines
8995 return null;
9096 }
9197 return (
92− <div key={phase.index}>
98+ <div key={`${run.taskId}:${phase.index}`}>
9399 <PhaseHeader phase={phase} />
94100 {rows.map((agent) => (
95− <div key={agent.index} className="px-0.5 py-0.5">
101+ <div key={`${run.taskId}:${agent.index}`} className="px-0.5 py-0.5">
96102 <AgentRowContent agent={agent} />
97103 </div>
98104 ))}
@@ -124,7 +130,7 @@ export function WorkflowRunCard(props: { · 25 unmodified lines
124130 }
125131
126132 function RemoteRunBody({ run }: { run: WorkflowRun }): ReactElement {
127− const sessionUrl = run.handles?.sessionUrl;
133+ const sessionUrl = safeWorkflowSessionUrl(run.handles?.sessionUrl);
128134 if (sessionUrl === undefined) {
129135 return (
130136 <p className="mt-1.5 px-0.5 text-[12px] text-muted-foreground/70 leading-5">
apps/web/src/components/workflow/workflowUi.tsx+20−1
@@ -157,8 +157,27 @@ export function AgentRowContent({ · 156 unmodified lines
157157 // Phase header + rollup helpers
158158 // ---------------------------------------------------------------------------
159159
160+/**
161+ * Only web URLs may reach an anchor href. The server already filters the
162+ * scheme at ingestion; this guards payloads persisted before that filter
163+ * (and any other producer) as defense in depth.
164+ */
165+export function safeWorkflowSessionUrl(sessionUrl: string | undefined): string | undefined {
166+ if (sessionUrl === undefined) {
167+ return undefined;
168+ }
169+ try {
170+ const parsed = new URL(sessionUrl);
171+ return parsed.protocol === "https:" || parsed.protocol === "http:" ? sessionUrl : undefined;
172+ } catch {
173+ return undefined;
174+ }
175+}
176+
177+/** Settled agents (done or error) — the x/y header is a progress counter,
178+ * and an errored agent has no work remaining. */
160179 export function phaseDoneCount(phase: WorkflowRunPhase): number {
161− return phase.agents.filter((agent) => agent.status === "done").length;
180+ return phase.agents.filter((agent) => agent.status === "done" || agent.status === "error").length;
162181 }
163182
164183 export function PhaseHeader({ phase }: { phase: WorkflowRunPhase }): ReactElement {
apps/web/src/workflow-logic.ts+31−3
@@ -300,13 +300,41 @@ export function collectWorkflowTaskIds( · 299 unmodified lines
300300 return taskIds;
301301 }
302302
303+/**
304+ * Millisecond timestamps collide, so equal-time activities are ordered by
305+ * provider sequence when present, then by lifecycle rank — a task.completed
306+ * must never be applied before the task.started that creates its run.
307+ */
308+const WORKFLOW_ACTIVITY_RANK: Record<string, number> = {
309+ "task.started": 0,
310+ "task.workflow-meta": 1,
311+ "task.workflow-updated": 1,
312+ "task.completed": 2,
313+};
314+
315+function compareWorkflowActivityOrder(
316+ a: OrchestrationThreadActivity,
317+ b: OrchestrationThreadActivity,
318+): number {
319+ if (a.sequence !== undefined && b.sequence !== undefined && a.sequence !== b.sequence) {
320+ return a.sequence - b.sequence;
321+ }
322+ const byTime = a.createdAt.localeCompare(b.createdAt);
323+ if (byTime !== 0) {
324+ return byTime;
325+ }
326+ const byRank = (WORKFLOW_ACTIVITY_RANK[a.kind] ?? 1) - (WORKFLOW_ACTIVITY_RANK[b.kind] ?? 1);
327+ if (byRank !== 0) {
328+ return byRank;
329+ }
330+ return a.id.localeCompare(b.id);
331+}
332+
303333 export function deriveWorkflowRuns(
304334 activities: ReadonlyArray<OrchestrationThreadActivity>,
305335 options?: { readonly sessionActive?: boolean | undefined },
306336 ): WorkflowRun[] {
307− const ordered = [...activities].toSorted(
308− (a, b) => a.createdAt.localeCompare(b.createdAt) || a.id.localeCompare(b.id),
309− );
337+ const ordered = [...activities].toSorted(compareWorkflowActivityOrder);
310338 const runs = new Map<string, MutableWorkflowRun>();
311339
312340 const ensureRun = (taskId: string, activity: OrchestrationThreadActivity): MutableWorkflowRun => {
packages/client-runtime/src/state/workflow.ts+4−0
@@ -25,6 +25,10 @@ export function createWorkflowEnvironmentAtoms<R, E>( · 24 unmodified lines
2525 label: "environment-data:workflow:read-journal",
2626 tag: WS_METHODS.workflowReadJournal,
2727 staleTimeMs: 5_000,
28+ // The journal grows while a run is live and the query only mounts
29+ // while the Logs tab is open — poll so new results appear without a
30+ // manual refresh.
31+ refreshIntervalMs: 4_000,
2832 }),
2933 readAgentTranscript: createEnvironmentRpcCommand(runtime, {
3034 label: "environment-data:workflow:read-agent-transcript",