groupthinking · Oct 27, 2026
@@ -0,0 +1,32 @@
+# Durable workflow task implementation — 2026-10-27
+
+Baseline: main `e93fd0e975cda0f3a59cb7fa812c19c64f6b2a52`.
+Authorization: Hayden requested action on the Oct 23 UVAI status report's recommended durable-task next step. This is a bounded implementation cut, not authorization to merge or change production.
+
+## Before and after
+
+`create_workflow_task` previously returned “Created task” with an in-memory object. It now requires a server-verified NextAuth provider/subject identity and a run ID, validates task inputs, writes through the existing Upstash REST credentials, and reads the persisted record before reporting fulfillment. It never falls back to memory. Storage errors return a failed action with no receipt.
+
+The task key is SHA-256 scoped to identity, run ID, and the fixed-order validated task fields. Atomic SET NX makes concurrent retries return the original task/creation time. Records do not expire. Task reads use the server-verified owner namespace, not client owner input. `GET /api/agents/actions?taskId=<receipt-id>` retrieves the same user's task. Receipt data survives action-agent serialization; the summary also includes its ID. Anonymous task writes now return 401 and missing run IDs return 400. Preview remains side-effect free.
+
+## Verification
+
+Commands executed locally with Node v24.22.1, npm 11.9.0 and installed Vitest 4.1.11:
+
+- Focused five-file suite: **42 passed, 0 failed, 0 skipped**. Files: workflow-task-store, action-tools, action-agent-routing, actions-route, action-agent-store.
+- Full web suite: **1,530 passed, 28 skipped, 171 files passed**, 31.05 seconds. This preceded the final provider/subject namespace tightening; the focused suite was rerun after that change and again passed all 42 tests.
+- `npm --prefix apps/web run type-check`: passed after the final code change.
+- Targeted ESLint for changed TypeScript files: passed; final identity files rechecked.
+- `git diff --check`: passed.
+
+The isolated REST fixture verifies 1 initial write + 3 concurrent retries produce 1 record and identical receipts. It also verifies owner isolation, run isolation, invalid-input rejection, signed-session verification, failure status, required readback, and receipt propagation. This fixture is not proof of live Upstash quality or restart durability.
+
+## Remaining acceptance work
+
+This is implementation evidence, not live video-to-action acceptance. No hosted store or paid AI provider was called by these tests. Successful real end-to-end runs: **not measured**. Cost/run and live latency: **unavailable**. No merge or production deployment was performed as part of implementation.
+
+The current reviewed-action transport supplies a run ID but not verified source timestamps or transcript/content hashes. Tasks therefore do not claim verified source provenance. `commitSha` records VERCEL_GIT_COMMIT_SHA when available, otherwise null. Source provenance, actual provider usage/cost, fresh-process hosted readback and authenticated UI acceptance remain required for the report's complete acceptance test. No scheduler, resource bookmark persistence or hosted execution adapter was added.
+
+Next acceptance exercise (requires an authorized non-production runtime/store and permitted source video): preview causes 0 task writes; confirm creates 1 task; three identical retries return the same ID with 1 record; a fresh authenticated process reads it; another owner cannot; a forced storage outage never yields fulfilled. Retain exact source URL, timestamps/content hash, run ID, build SHA and actual usage/cost or explicitly unknown values. Do not mark complete until those receipts exist.
+
+Related PR #2470 adds a local review-only journal and expressly excludes live action execution. This cut does not duplicate that journal. The two PRs touch the action route and will need integration review before either is merged.
@@ -1,5 +1,7 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 
+vi.mock('@/lib/workflow-task-store', () => ({ workflowTaskOwner: vi.fn(async () => 'user-1'), readWorkflowTask: vi.fn() }));
+
 vi.mock('@/lib/action-agent', () => ({
   AVAILABLE_TOOL_NAMES: ['create_workflow_task', 'dispatch_agent'],
   runActionAgent: vi.fn(),
@@ -13,7 +15,8 @@ vi.mock('@/lib/billing/entitlement-store', () => ({
   isProSubscriber: vi.fn(async () => false),
 }));
 
-import { POST } from '@/app/api/agents/actions/route';
+import { workflowTaskOwner, readWorkflowTask } from '@/lib/workflow-task-store';
+import { GET, POST } from '@/app/api/agents/actions/route';
 import { executePreparedActions, runActionAgent } from '@/lib/action-agent';
 import { isProSubscriber } from '@/lib/billing/entitlement-store';
 
@@ -32,6 +35,7 @@ function request(body: unknown): Request {
 describe('POST /api/agents/actions review gate', () => {
   beforeEach(() => {
     vi.clearAllMocks();
+    vi.mocked(workflowTaskOwner).mockResolvedValue('user-1');
     mockedIsProSubscriber.mockResolvedValue(false);
   });
 
@@ -71,6 +75,7 @@ describe('POST /api/agents/actions review gate', () => {
     expect(mockedExecutePreparedActions).toHaveBeenCalledWith({
       actions: [prepared],
       jobId: 'job_1',
+      ownerId: 'user-1',
     });
     expect(mockedRunActionAgent).not.toHaveBeenCalled();
   });
@@ -111,4 +116,27 @@ describe('POST /api/agents/actions review gate', () => {
     expect((await POST(request({ mode: 'execute', actions: [] }))).status).toBe(400);
     expect((await POST(request({ mode: 'surprise' }))).status).toBe(400);
   });
+  it('rejects unauthenticated task writes even with a supplied owner', async () => {
+    vi.mocked(workflowTaskOwner).mockResolvedValue(null);
+    const response = await POST(request({ mode: 'execute', ownerId: 'spoof', jobId: 'run', actions: [{ tool: 'create_workflow_task', input: {} }] }));
+    expect(response.status).toBe(401);
+    expect(mockedExecutePreparedActions).not.toHaveBeenCalled();
+  });
+
+  it('requires a stable run ID for task execution', async () => {
+    const response = await POST(request({ mode: 'execute', actions: [{ tool: 'create_workflow_task', input: {} }] }));
+    expect(response.status).toBe(400);
+    expect(mockedExecutePreparedActions).not.toHaveBeenCalled();
+  });
+
+  it('reads only the current session owner namespace', async () => {
+    const id = 'a'.repeat(64);
+    vi.mocked(readWorkflowTask).mockResolvedValue(null);
+    const response = await GET(new Request(`http://localhost/api/agents/actions?taskId=${id}&ownerId=other`));
+    expect(response.status).toBe(404);
+    expect(readWorkflowTask).toHaveBeenCalledWith('user-1', id);
+    vi.mocked(workflowTaskOwner).mockResolvedValue(null);
+    expect((await GET(new Request(`http://localhost/api/agents/actions?taskId=${id}`))).status).toBe(401);
+  });
+
 });
@@ -1,3 +1,4 @@
+import { workflowTaskOwner, readWorkflowTask } from '@/lib/workflow-task-store';
 import { NextResponse } from 'next/server';
 import { executePreparedActions, runActionAgent } from '@/lib/action-agent';
 import { AVAILABLE_TOOL_NAMES } from '@/lib/action-agent';
@@ -31,7 +32,17 @@ function isPreparedAction(value: unknown): value is AgentAction {
  * Advertises the executable tools the action agent can invoke, so the UI can
  * render affordances without hardcoding the list.
  */
-export async function GET(): Promise<NextResponse> {
+export async function GET(request?: Request): Promise<NextResponse> {
+  const taskId = request && new URL(request.url).searchParams.get('taskId');
+  if (taskId && request) {
+    const owner = await workflowTaskOwner(request);
+    if (!owner) return NextResponse.json({ error: 'Sign in to read tasks' }, { status: 401 });
+    if (!/^[a-f0-9]{64}$/.test(taskId)) return NextResponse.json({ error: 'Invalid task ID' }, { status: 400 });
+    try {
+      const task = await readWorkflowTask(owner, taskId);
+      return NextResponse.json(task ? { task } : { error: 'Task not found' }, { status: task ? 200 : 404 });
+    } catch { return NextResponse.json({ error: 'Task storage unavailable' }, { status: 503 }); }
+  }
   const hasKey = !!(process.env.OPENAI_API_KEY || hasGeminiKey());
   return NextResponse.json({ available: hasKey, tools: AVAILABLE_TOOL_NAMES });
 }
@@ -91,6 +102,13 @@ export async function POST(request: Request): Promise<NextResponse> {
       );
     }
 
+    let ownerId: string | undefined;
+    if (reviewedActions.some(action => action.tool === 'create_workflow_task')) {
+      ownerId = (await workflowTaskOwner(request)) ?? undefined;
+      if (!ownerId) return NextResponse.json({ success: false, error: 'Sign in to save tasks', actions: [] }, { status: 401 });
+      if (typeof jobId !== 'string' || !jobId.trim() || jobId.length > 200) return NextResponse.json({ success: false, error: 'A run ID is required', actions: [] }, { status: 400 });
+    }
+
     if (reviewedActions.some((action) => PRO_ACTION_TOOLS.has(action.tool))) {
       const billingEmail = await resolveTrustedBillingEmail(request);
       if (!(await isProSubscriber(billingEmail))) {
@@ -109,6 +127,7 @@ export async function POST(request: Request): Promise<NextResponse> {
     try {
       const fulfilled = await executePreparedActions({
         actions: reviewedActions,
+        ownerId,
         jobId: typeof jobId === 'string' ? jobId : undefined,
       });
       return NextResponse.json({ success: true, provider: 'confirmed-plan', actions: fulfilled });

@@ -44,15 +44,14 @@ describe('action tool registry', () => {
     expect(getTool('nope')).toBeUndefined();
   });
 
-  it('create_workflow_task produces real structured output (no fabrication)', async () => {
+  it('create_workflow_task fails closed without an authenticated owner', async () => {
     const tool = getTool('create_workflow_task')!;
     const res = await tool.execute(
       { title: 'Install pnpm', description: 'Set up the toolchain', category: 'setup', priority: 'high' },
       NO_BACKEND,
     );
-    expect(res.isError).toBeFalsy();
-    expect(res.summary).toContain('Install pnpm');
-    expect(res.data).toMatchObject({ title: 'Install pnpm', category: 'setup', priority: 'high' });
+    expect(res.isError).toBe(true);
+    expect(res.data).toBeUndefined();
   });
 
   it('dispatch_agent reports honestly when no backend is configured', async () => {

@@ -0,0 +1,65 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import { encode } from 'next-auth/jwt';
+import { createWorkflowTaskRecord, readWorkflowTask, workflowTaskOwner } from '../workflow-task-store';
+import { executePreparedActions } from '../action-agent';
+
+const input = { title: 'Install pnpm', description: 'Set up the toolchain', category: 'setup', priority: 'high' };
+const records = new Map<string, string>();
+const transport = vi.fn(async (_url: unknown, init?: RequestInit) => {
+  const [op, key, value] = JSON.parse(String(init?.body));
+  let result: unknown = null;
+  if (op === 'SET' && !records.has(key)) { records.set(key, value); result = 'OK'; }
+  if (op === 'GET') result = records.get(key) ?? null;
+  return Response.json({ result });
+}) as unknown as typeof fetch;
+
+describe('durable workflow task contract (isolated REST fixture)', () => {
+  beforeEach(() => {
+    records.clear(); vi.clearAllMocks();
+    vi.stubEnv('UPSTASH_REDIS_REST_URL', 'https://isolated.invalid');
+    vi.stubEnv('UPSTASH_REDIS_REST_TOKEN', 'fixture');
+  });
+  afterEach(() => vi.unstubAllEnvs());
+  it('confirms one record and returns the same receipt for three concurrent retries', async () => {
+    const first = await createWorkflowTaskRecord(input, 'alice', 'run-1', transport);
+    const retries = await Promise.all(Array.from({ length: 3 }, () => createWorkflowTaskRecord(input, 'alice', 'run-1', transport)));
+    expect(records.size).toBe(1);
+    expect(retries).toEqual([first, first, first]);
+    expect(await readWorkflowTask('alice', first.id, transport)).toEqual(first);
+    expect(await readWorkflowTask('bob', first.id, transport)).toBeNull();
+  });
+  it('scopes identical actions to owner and run', async () => {
+    const a = await createWorkflowTaskRecord(input, 'alice', 'run-1', transport);
+    const b = await createWorkflowTaskRecord(input, 'bob', 'run-1', transport);
+    const c = await createWorkflowTaskRecord(input, 'alice', 'run-2', transport);
+    expect(new Set([a.id, b.id, c.id]).size).toBe(3);
+  });
+  it('rejects invalid inputs and missing identity before storage', async () => {
+    for (const args of [[{}, 'alice', 'run'], [input, undefined, 'run'], [input, 'alice', undefined]] as const) {
+      await expect(createWorkflowTaskRecord(args[0], args[1], args[2], transport)).rejects.toThrow();
+    }
+    expect(transport).not.toHaveBeenCalled();
+  });
+  it('never returns fulfilled on storage failure', async () => {
+    const actions = await executePreparedActions({ actions: [{ tool: 'create_workflow_task', input, status: 'pending' }], ownerId: 'alice', jobId: 'run', fetchImpl: async () => new Response('unavailable', { status: 503 }) });
+    expect(actions[0].status).toBe('failed'); expect(actions[0].data).toBeUndefined();
+  });
+  it('requires readback after a successful write acknowledgement', async () => {
+    let call = 0;
+    await expect(createWorkflowTaskRecord(input, 'alice', 'run', async () => Response.json({ result: ++call === 1 ? 'OK' : null }))).rejects.toThrow('readback');
+  });
+  it('preserves the durable ID through action execution', async () => {
+    const actions = await executePreparedActions({ actions: [{ tool: 'create_workflow_task', input, status: 'pending' }], ownerId: 'alice', jobId: 'run', fetchImpl: transport });
+    expect(actions[0].status).toBe('fulfilled');
+    expect(actions[0].data?.id).toMatch(/^[a-f0-9]{64}$/);
+  });
+  it('accepts signed session identity and rejects client identity and forged cookies', async () => {
+    vi.stubEnv('NEXTAUTH_SECRET', 'isolated-test-secret');
+    vi.stubEnv('NODE_ENV', 'test');
+    vi.stubEnv('NEXTAUTH_URL', 'http://localhost');
+    const token = await encode({ token: { sub: 'alice', provider: 'github' }, secret: 'isolated-test-secret' });
+    expect(await workflowTaskOwner(new Request('http://localhost', { headers: { cookie: `next-auth.session-token=${token}` } }))).toBe(JSON.stringify(['github', 'alice']));
+    expect(await workflowTaskOwner(new Request('http://localhost', { headers: { 'x-user-id': 'alice', cookie: 'next-auth.session-token=forged' } }))).toBeNull();
+  });
+
+});
@@ -96,6 +96,7 @@ function toAction(
     input,
     status: status ?? (result.isError ? 'failed' : 'fulfilled'),
     result: result.summary,
+    ...(tool === 'create_workflow_task' && result.data ? { data: result.data } : {}),
     isError: result.isError,
   };
 }
@@ -302,6 +303,7 @@ export async function runActionAgent(opts: RunActionAgentOptions): Promise<RunAc
 }
 
 export interface ExecutePreparedActionsOptions {
+  ownerId?: string;
   actions: AgentAction[];
   jobId?: string;
   fetchImpl?: typeof fetch;
@@ -315,6 +317,7 @@ export async function executePreparedActions(
     backendBaseUrl: resolveBackendBaseUrl(),
     fetchImpl: opts.fetchImpl,
     jobId: opts.jobId,
+    ownerId: opts.ownerId,
   };
 
   const uniqueActions = dedupeAgentActions(opts.actions);

@@ -33,6 +33,8 @@ export interface AgentAction {
   status: 'pending' | 'fulfilled' | 'failed';
   /** Human-readable result summary once executed. */
   result?: string;
+  /** Durable tool receipt, when available. */
+  data?: Record<string, unknown>;
   /** Whether the underlying tool reported an error. */
   isError?: boolean;
 }

@@ -1,3 +1,4 @@
+import { createWorkflowTaskRecord } from './workflow-task-store';
 /**
  * Executable tool registry for the Transcription-Driven Action Agent.
  *
@@ -37,6 +38,8 @@ export interface ActionToolResult {
 
 /** Runtime context handed to every tool execution. */
 export interface ToolContext {
+  /** Server-verified identity, only supplied after explicit confirmation. */
+  ownerId?: string;
   /** Resolved FastAPI base URL, or null when no backend is configured. */
   backendBaseUrl: string | null;
   /** Injectable fetch (defaults to global fetch) so tools are testable. */
@@ -96,18 +99,13 @@ const createWorkflowTask: ActionTool = {
     required: ['title', 'description', 'category', 'priority'],
     additionalProperties: false,
   },
-  async execute(input) {
-    const title = str(input, 'title');
-    return {
-      summary: `Created task "${title}"`,
-      data: {
-        title,
-        description: str(input, 'description'),
-        category: str(input, 'category'),
-        priority: str(input, 'priority'),
-        createdAt: new Date().toISOString(),
-      },
-    };
+  async execute(input, ctx) {
+    try {
+      const task = await createWorkflowTaskRecord(input, ctx.ownerId, ctx.jobId, ctx.fetchImpl);
+      return { summary: `Saved task "${task.title}" (${task.id})`, data: task };
+    } catch {
+      return { summary: 'Task was not confirmed saved. Sign in, supply a run ID, and retry when durable storage is available.', isError: true };
+    }
   },
 };