UX — F75 workbench active state, F95, F97 proof of life on the harness card

apps/aflow-executor-host/src/__tests__/harnessEvents.test.ts

@@ -241,3 +241,94 @@ describe('what a tool call reads as', () => {
     });
   });
 });
+
+/**
+ * Proof of life: what the agent says it is doing, the files it changes, and
+ * which results are a command's exit status — read off the fields of each call,
+ * so a renamed tool costs nothing.
+ */
+describe('what a watcher reads first', () => {
+  function readCalls(
+    calls: Array<{ id: string; name: string; input: unknown }>,
+    results: Array<{ id: string; isError: boolean }> = [],
+  ): HarnessActivityLine[] {
+    const lines: HarnessActivityLine[] = [];
+    const reader = createHarnessEventReader('claude-stream-json', (l) => lines.push(l), 0);
+    for (const call of calls) {
+      reader.push(
+        `${JSON.stringify({
+          type: 'assistant',
+          message: { content: [{ type: 'tool_use', ...call }] },
+        })}\n`,
+      );
+    }
+    for (const result of results) {
+      reader.push(
+        `${JSON.stringify({
+          type: 'user',
+          message: {
+            content: [
+              {
+                type: 'tool_result',
+                tool_use_id: result.id,
+                content: 'out',
+                is_error: result.isError,
+              },
+            ],
+          },
+        })}\n`,
+      );
+    }
+    return lines;
+  }
+
+  it('names the file a call changes, and not one it only reads', () => {
+    const lines = readCalls([
+      { id: 'a', name: 'Edit', input: { file_path: 'src/a.ts', old_string: 'x', new_string: 'y' } },
+      { id: 'b', name: 'Write', input: { file_path: 'src/b.ts', content: 'z' } },
+      { id: 'c', name: 'Read', input: { file_path: 'src/c.ts' } },
+      { id: 'd', name: 'Grep', input: { pattern: 'parse', path: 'src' } },
+    ]);
+    expect(lines.map((l) => (l.kind === 'tool' ? (l.changes ?? null) : 'other'))).toEqual([
+      'src/a.ts',
+      'src/b.ts',
+      null,
+      null,
+    ]);
+  });
+
+  it('says what the agent marked in progress, once per change of plan', () => {
+    const plan = (active: string) => ({
+      todos: [
+        { content: 'Read the parser', status: 'completed', activeForm: 'Reading the parser' },
+        { content: active, status: 'in_progress', activeForm: `${active}…` },
+      ],
+    });
+    const lines = readCalls([
+      { id: 'a', name: 'TodoWrite', input: plan('Run the tests') },
+      { id: 'b', name: 'TodoWrite', input: plan('Run the tests') },
+      { id: 'c', name: 'TodoWrite', input: plan('Fix the guard') },
+    ]);
+    expect(lines.filter((l) => l.kind === 'intent').map((l) => l.text)).toEqual([
+      'Run the tests…',
+      'Fix the guard…',
+    ]);
+  });
+
+  it("marks a command's result as one, so its exit status can be judged apart", () => {
+    const lines = readCalls(
+      [
+        { id: 'a', name: 'Bash', input: { command: 'yarn test' } },
+        { id: 'b', name: 'Read', input: { file_path: 'missing.ts' } },
+      ],
+      [
+        { id: 'a', isError: true },
+        { id: 'b', isError: true },
+      ],
+    );
+    const results = lines.filter((l) => l.kind === 'tool_result');
+    expect(results[0]).toMatchObject({ tool: 'Bash', ok: false, command: true });
+    expect(results[1]).toMatchObject({ tool: 'Read', ok: false });
+    expect(results[1]).not.toHaveProperty('command');
+  });
+});

apps/aflow-executor-host/src/harnessEvents.ts

@@ -83,6 +83,50 @@ export function summarizeToolUse(name: string, input: unknown): string {
   return name;
 }
 
+/**
+ * The file a call changes, when it changes one.
+ *
+ * Fields again, not names: a call that names a path and carries what to put
+ * there is a change, whatever the harness calls the tool. A call naming a path
+ * alone is a read.
+ */
+const PATH_FIELDS = ['file_path', 'notebook_path', 'path'] as const;
+const CONTENT_FIELDS = ['content', 'new_string', 'edits', 'new_source'] as const;
+
+export function changedFileOf(input: unknown): string | undefined {
+  if (!isRecord(input)) return undefined;
+  if (!CONTENT_FIELDS.some((field) => input[field] !== undefined)) return undefined;
+  for (const field of PATH_FIELDS) {
+    const value = input[field];
+    if (typeof value === 'string' && value.trim() !== '') return clip(value, SUMMARY_MAX);
+  }
+  return undefined;
+}
+
+/** A call that hands the shell a command line. */
+export function isCommandCall(input: unknown): boolean {
+  return isRecord(input) && typeof input['command'] === 'string';
+}
+
+/**
+ * What the agent says it is doing: the item of its plan it marked in progress.
+ *
+ * A harness that keeps a plan rewrites the whole of it on every change, so the
+ * call carries every item and the one in progress is the news. Its present-tense
+ * wording is preferred where the harness writes one.
+ */
+export function intentOf(input: unknown): string | undefined {
+  if (!isRecord(input) || !Array.isArray(input['todos'])) return undefined;
+  for (const item of input['todos']) {
+    if (!isRecord(item) || item['status'] !== 'in_progress') continue;
+    for (const field of ['activeForm', 'content'] as const) {
+      const value = item[field];
+      if (typeof value === 'string' && value.trim() !== '') return clip(value, SUMMARY_MAX);
+    }
+  }
+  return undefined;
+}
+
 /** A tool result arrives as a string or as content blocks; both read as text. */
 function textOfContent(content: unknown): string {
   if (typeof content === 'string') return content;
@@ -125,8 +169,10 @@ export function createHarnessEventReader(
   const passedThrough: string[] = [];
   // A tool result names the call it answers, not the tool that made it, so the
   // name is carried across from the call. Without it a feed says a result came
-  // back and never which of five in-flight calls it belongs to.
-  const toolNames = new Map<string, string>();
+  // back and never which of five in-flight calls it belongs to. Whether the
+  // call ran a command travels with it for the same reason.
+  const toolCalls = new Map<string, { name: string; command: boolean }>();
+  let lastIntent = '';
 
   const at = (): number => Math.max(0, Date.now() - startedAt);
 
@@ -152,14 +198,24 @@ export function createHarnessEventReader(
       }
       if (block['type'] === 'tool_use' && typeof block['name'] === 'string') {
         const name = block['name'];
+        const input = block['input'];
         const id = block['id'];
-        if (typeof id === 'string') toolNames.set(id, name);
+        if (typeof id === 'string') toolCalls.set(id, { name, command: isCommandCall(input) });
+        const changes = changedFileOf(input);
         onLine({
           kind: 'tool',
           at: at(),
           tool: name,
-          text: summarizeToolUse(name, block['input']),
+          text: summarizeToolUse(name, input),
+          ...(changes !== undefined ? { changes } : {}),
         });
+        // A plan rewritten without moving on says nothing new, and repeating
+        // the same intent would date it as fresh news.
+        const intent = intentOf(input);
+        if (intent !== undefined && intent !== lastIntent) {
+          lastIntent = intent;
+          onLine({ kind: 'intent', at: at(), text: intent });
+        }
       }
     }
   };
@@ -170,13 +226,14 @@ export function createHarnessEventReader(
     for (const block of content) {
       if (!isRecord(block) || block['type'] !== 'tool_result') continue;
       const id = block['tool_use_id'];
-      const tool = (typeof id === 'string' ? toolNames.get(id) : undefined) ?? 'tool';
+      const call = typeof id === 'string' ? toolCalls.get(id) : undefined;
       onLine({
         kind: 'tool_result',
         at: at(),
-        tool,
+        tool: call?.name ?? 'tool',
         ok: block['is_error'] !== true,
         text: summarizeToolResult(block['content']),
+        ...(call?.command === true ? { command: true } : {}),
       });
     }
   };

apps/aflow-orchestrator/src/services/ProjectionWorker.ts

@@ -1,3 +1,4 @@
+import { randomUUID } from 'node:crypto';
 import type { Redis } from 'ioredis';
 import type postgres from 'postgres';
 import type { PostgresJsDatabase } from 'drizzle-orm/postgres-js';
@@ -16,6 +17,7 @@ import {
   compareStreamIds,
   deleteRecoveryData,
   publishActionCenterWake,
+  appendEntityEvent,
   type ProjectionCandidate,
   type SessionHotState,
   type SessionEvent,
@@ -145,6 +147,17 @@ const EVENT_FLUSH_PAGE = 1000;
 
 const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
 
+/**
+ * Whether a persisted event moved the session between statuses — a turn
+ * beginning, resting, resuming or ending. Every such event is a `Session*`
+ * lifecycle event; step and run events move nothing a conversation list shows.
+ */
+export function movesSessionStatus(
+  event: Pick<SessionEvent, 'eventType'> | null | undefined,
+): boolean {
+  return event?.eventType.startsWith('Session') === true;
+}
+
 function isUuid(value: string): boolean {
   return UUID_RE.test(value);
 }
@@ -278,7 +291,7 @@ export function createProjectionWorker(
     tx: PostgresJsDatabase,
     tenantId: string,
     runId: string,
-  ): Promise<{ evictedGap: boolean }> {
+  ): Promise<{ evictedGap: boolean; statusMoved: boolean }> {
     const [row] = await tx
       .select({ cursor: sessions.lastFlushedEventStreamId, status: sessions.status })
       .from(sessions)
@@ -297,6 +310,7 @@ export function createProjectionWorker(
     let cursor = initialCursor;
     const now = new Date();
     let evictedGap = false;
+    let statusMoved = false;
 
     for (;;) {
       const { entries, lastId, oldestId } = await readDurableEventEntries(
@@ -340,6 +354,7 @@ export function createProjectionWorker(
         }
       }
       if (entries.length > 0) {
+        if (!statusMoved) statusMoved = entries.some((entry) => movesSessionStatus(entry.event));
         await tx
           .insert(eventLog)
           .values(
@@ -359,7 +374,46 @@ export function createProjectionWorker(
         .set({ lastFlushedEventStreamId: cursor })
         .where(eq(sessions.sessionId, runId));
     }
-    return { evictedGap };
+    return { evictedGap, statusMoved };
+  }
+
+  /**
+   * Tell the space a conversation's status moved, once the move is durable.
+   *
+   * Driven by the lifecycle events this projection just persisted rather than
+   * by the projection itself, which runs on every step: a turn of forty tool
+   * calls is two signals — it began, it rested — not forty. Top-level sessions
+   * only, because those are the conversations a list shows; a child's turn is
+   * its parent's. Best-effort: the list refetches on its own staleness.
+   */
+  async function publishSessionStatus(
+    tenantId: string,
+    runId: string,
+    runState: SessionHotState,
+  ): Promise<void> {
+    if (!runState.spaceId || runState.parentSessionId) return;
+    try {
+      await appendEntityEvent(redis, {
+        tenantId,
+        spaceId: runState.spaceId,
+        event: {
+          eventId: randomUUID(),
+          eventType: 'entity.session.status',
+          spaceId: runState.spaceId,
+          tenantId,
+          timestamp: Date.now(),
+          causedBySessionId: runId,
+          payload: { sessionId: runId, status: runState.status },
+          summary: `Conversation ${runState.status.toLowerCase()}`,
+        },
+      });
+    } catch (err) {
+      log.warn(`[ProjectionWorker] session status signal failed for run ${runId}`, {
+        tenantId,
+        runId,
+        error: err instanceof Error ? err.message : String(err),
+      });
+    }
   }
 
   // ── Full flush (terminal runs) ──────────────────────────────────────────
@@ -378,7 +432,7 @@ export function createProjectionWorker(
     // all, and the loss is silent until someone tries.
     const hotStateSnapshot = await buildHotStateSnapshot(redis, tenantId, runState);
     let fenceWon = false;
-    await withTenantSchema(db, tenantContext, async (tx) => {
+    const statusMoved = await withTenantSchema(db, tenantContext, async (tx) => {
       const upserted = await tx
         .insert(sessions)
         .values({
@@ -438,7 +492,8 @@ export function createProjectionWorker(
         })
         .returning({ sessionId: sessions.sessionId });
       fenceWon = upserted.length > 0;
-      if (fenceWon) await flushDurableEvents(tx, tenantId, runId);
+      if (!fenceWon) return false;
+      return (await flushDurableEvents(tx, tenantId, runId)).statusMoved;
     });
 
     if (!fenceWon) {
@@ -472,6 +527,7 @@ export function createProjectionWorker(
         await bumpAttentionGeneration(redis, tenantId, runState.spaceId);
       }
     }
+    if (statusMoved) await publishSessionStatus(tenantId, runId, runState);
 
     // After the upsert, because the mark it claims lives on the row that upsert
     // writes; not swallowed, because a throw here leaves the candidate
@@ -617,9 +673,10 @@ export function createProjectionWorker(
     // from the status upsert, so a failing insert leaves the status row
     // current — the throw still withholds the acknowledgement, which is what
     // re-drives the events.
-    await withTenantSchema(db, tenantContext, async (tx) => {
-      await flushDurableEvents(tx, tenantId, runId);
-    });
+    const { statusMoved } = await withTenantSchema(db, tenantContext, (tx) =>
+      flushDurableEvents(tx, tenantId, runId),
+    );
+    if (statusMoved) await publishSessionStatus(tenantId, runId, runState);
 
     // The candidate is acknowledged by the caller; a later mutation re-arms it.
     // Leaving it armed instead would re-upsert every running session every