feat(agentic): add @agentic-kit/run-log-gate — tool approvals over the run log

.github/workflows/run-tests.yaml

@@ -115,7 +115,7 @@ jobs:
           - batch: graphile-unit
             packages: 'graphile/graphile-plugin-utils graphile/graphile-realtime-subscriptions graphile/graphile-sql-expression-validator graphile/graphile-upload-plugin graphile/graphile-storage-registry'
           - batch: agentic
-            packages: 'agentic/protocol agentic/agentic-kit agentic/agent agentic/harness agentic/chat agentic/cli agentic/db-tools agentic/pi agentic/dsh agentic/react agentic/agentic-server agentic/anthropic agentic/openai agentic/ollama agentic/run-log agentic/metering agentic/agent-conversation agentic/pi-host agentic/run-log-client'
+            packages: 'agentic/protocol agentic/agentic-kit agentic/agent agentic/harness agentic/chat agentic/cli agentic/db-tools agentic/pi agentic/dsh agentic/react agentic/agentic-server agentic/anthropic agentic/openai agentic/ollama agentic/run-log agentic/metering agentic/agent-conversation agentic/pi-host agentic/run-log-client agentic/run-log-gate'
           - batch: pgpm-unit
             packages: 'pgpm/types pgpm/naming-spec pgpm/diff pgpm/import pgpm/slice pgpm/transform'
           - batch: pglite

agentic/run-log-gate/README.md

@@ -0,0 +1,39 @@
+# @agentic-kit/run-log-gate
+
+Tool approvals over the run log. The request, the human's answer and every
+settled gate decision are entries in the run's log — `agent_event` rows once
+the store is the platform's — so a desktop run, a cloud pi Job and the agentic
+lane all ask in one place and are answered from any surface that projects the
+log (`RunsPanel`, `pendingApprovals` → `resolveApproval`).
+
+## Three pieces
+
+- `runLogApprovals` — an `ApprovalChannel` for `@agentic-kit/harness`'s
+  `RunGate`. Submits a `constructive.approval.request` entry and polls the log
+  for the matching `constructive.approval.resolution`. The request entry's id
+  is derived from the tool call id, so a retried submit is the same entry and
+  the store's idempotency drops it instead of asking a human twice.
+  `resolveApproval` is the other end for a local run and the tests.
+- `runLogGateDecisions` — an `onDecision` sink that appends every settled
+  decision as a `constructive.gate.decision` entry (audit: which calls the gate
+  refused, on whose authority) and a `flush()` that rethrows the first write
+  that failed.
+- `createRunLogGateHost` — the harness's `GateHost` (what `createConfirmGate`
+  and `createGatedToolset` drive) built on the two above: `confirmTool` asks in
+  the log and returns what the log settled on, `notifyToolSkipped` files the
+  skipped repeat as a denial, `drain()` surfaces a lost write.
+
+Nothing here executes on the strength of a click: the decision is the record
+that was persisted and read back. A host may mirror a concise notice into a
+conversation thread, but the thread is never the authority.
+
+## Consumers
+
+In constructive-db: `pi-job-host` (the cloud pi Job — `RunGate` over
+`runLogApprovals` + `runLogGateDecisions`), the `code_task` agentic lane and
+`ui/desktop`'s capability gate (`createRunLogGateHost`).
+
+## Testing
+
+`pnpm test` — the store (`MemoryRunLogStore`), the clock and the sleep are all
+injectable, so every path is covered without a database or a network.

agentic/run-log-gate/__tests__/approvals.test.ts

@@ -0,0 +1,154 @@
+import type { ApprovalRequest } from '@agentic-kit/harness';
+import type { PiSessionEntry } from '@agentic-kit/run-log';
+import { MemoryRunLogStore, projectToolState, readAll } from '@agentic-kit/run-log';
+
+import { approvalEntryId, resolveApproval, runLogApprovals } from '../src/approvals';
+
+const request = (overrides: Partial<ApprovalRequest> = {}): ApprovalRequest => ({
+  runId: 'run-1',
+  toolCallId: 'call-1',
+  toolName: 'bash',
+  input: { command: 'rm -rf /' },
+  reason: 'policy asks about destructive commands',
+  requestedAt: '2024-01-01T00:00:00.000Z',
+  ...overrides
+});
+
+/** Let every pending read/append settle — the submit is several awaits deep. */
+const flush = (): Promise<void> => new Promise((resolve) => setImmediate(resolve));
+
+const message = (id: string, text: string): PiSessionEntry =>
+  ({
+    type: 'message',
+    id,
+    parentId: null,
+    timestamp: '2024-01-01T00:00:00Z',
+    message: { role: 'assistant', content: [{ type: 'text', text }] }
+  }) as PiSessionEntry;
+
+describe('runLogApprovals', () => {
+  let store: MemoryRunLogStore;
+
+  beforeEach(() => {
+    store = new MemoryRunLogStore();
+  });
+
+  const channel = (overrides: Partial<Parameters<typeof runLogApprovals>[0]> = {}) =>
+    runLogApprovals({
+      store,
+      runId: 'run-1',
+      intervalMs: 0,
+      // A macrotask rather than a microtask: a poll loop that only ever yields to
+      // the microtask queue starves the run's timers for as long as it spins.
+      sleep: () => new Promise<void>((resolve) => setImmediate(resolve)),
+      now: () => new Date('2024-01-01T00:00:00.000Z'),
+      ...overrides
+    });
+
+  it('writes the request into the log where a projector finds it pending', async () => {
+    const pending = channel().request(request());
+    // Let the submit land before answering, the way a human would.
+    await flush();
+
+    const projected = projectToolState(await readAll(store, 'run-1'));
+    expect(projected.pendingApprovals).toEqual([
+      expect.objectContaining({
+        toolCallId: 'call-1',
+        prompt: 'bash: policy asks about destructive commands'
+      })
+    ]);
+
+    await resolveApproval(store, 'run-1', { toolCallId: 'call-1', decision: 'approved' });
+    await expect(pending).resolves.toEqual({ decision: 'allow' });
+  });
+
+  it('carries the resolution\'s reason and actor back to the gate', async () => {
+    const pending = channel().request(request());
+    await flush();
+    await resolveApproval(store, 'run-1', {
+      toolCallId: 'call-1',
+      decision: 'rejected',
+      reason: 'not on production',
+      actorId: 'user-7'
+    });
+
+    await expect(pending).resolves.toEqual({
+      decision: 'deny',
+      reason: 'not on production',
+      actorId: 'user-7'
+    });
+  });
+
+  it('sees a resolution appended after unrelated entries in a later page', async () => {
+    const pending = channel({ pageLimit: 1 }).request(request());
+    await flush();
+    await store.append('run-1', [message('chatter-1', 'still thinking')]);
+    await store.append('run-1', [message('chatter-2', 'still thinking')]);
+    await resolveApproval(store, 'run-1', { toolCallId: 'call-1', decision: 'approved' });
+
+    await expect(pending).resolves.toEqual({ decision: 'allow' });
+  });
+
+  it('asks once when a submit is retried, because the entry id is the tool call', async () => {
+    const first = channel().request(request());
+    await flush();
+    const second = channel().request(request());
+    await flush();
+
+    await resolveApproval(store, 'run-1', { toolCallId: 'call-1', decision: 'approved' });
+    await expect(first).resolves.toEqual({ decision: 'allow' });
+    await expect(second).resolves.toEqual({ decision: 'allow' });
+
+    const records = await readAll(store, 'run-1');
+    const requests = records.filter(
+      (candidate) => (candidate.entry as { id?: string }).id === approvalEntryId('call-1')
+    );
+    expect(requests).toHaveLength(1);
+  });
+
+  it('hangs the request off the log\'s leaf so the session stays one tree', async () => {
+    await store.append('run-1', [message('leaf', 'about to run a tool')]);
+    // Bounded so the request settles instead of polling past the assertion.
+    const pending = channel({ timeoutMs: 0 }).request(request());
+    await pending;
+
+    const records = await readAll(store, 'run-1');
+    const submitted = records[records.length - 1].entry as { parentId?: string | null };
+    expect(submitted.parentId).toBe('leaf');
+  });
+
+  it('denies on timeout rather than proceeding unapproved', async () => {
+    let clock = 0;
+    const outcome = await runLogApprovals({
+      store,
+      runId: 'run-1',
+      intervalMs: 1,
+      timeoutMs: 10,
+      sleep: () => {
+        clock += 10;
+        return Promise.resolve();
+      },
+      now: () => new Date(clock)
+    }).request(request());
+
+    expect(outcome).toEqual({ decision: 'deny', reason: 'gate: no decision within 10ms' });
+  });
+
+  it('can be told to allow on timeout', async () => {
+    let clock = 0;
+    const outcome = await runLogApprovals({
+      store,
+      runId: 'run-1',
+      intervalMs: 1,
+      timeoutMs: 10,
+      onTimeout: 'allow',
+      sleep: () => {
+        clock += 10;
+        return Promise.resolve();
+      },
+      now: () => new Date(clock)
+    }).request(request());
+
+    expect(outcome.decision).toBe('allow');
+  });
+});

agentic/run-log-gate/__tests__/gate-decisions.test.ts

@@ -0,0 +1,119 @@
+import type { RunGateDecisionRecord } from '@agentic-kit/harness';
+import type { PiSessionEntry } from '@agentic-kit/run-log';
+import { MemoryRunLogStore, projectToolState, readAll } from '@agentic-kit/run-log';
+
+import { gateDecisionEntryId, runLogGateDecisions } from '../src/gate-decisions';
+
+const decision = (overrides: Partial<RunGateDecisionRecord> = {}): RunGateDecisionRecord => ({
+  runId: 'run-1',
+  toolCallId: 'call-1',
+  toolName: 'bash',
+  input: { command: 'rm -rf /' },
+  verdict: { decision: 'deny', reason: 'destructive commands are not permitted' },
+  decision: 'deny',
+  reason: 'destructive commands are not permitted',
+  decidedAt: '2024-01-01T00:00:00.000Z',
+  ...overrides
+});
+
+const message = (id: string, text: string): PiSessionEntry =>
+  ({
+    type: 'message',
+    id,
+    parentId: null,
+    timestamp: '2024-01-01T00:00:00Z',
+    message: { role: 'assistant', content: [{ type: 'text', text }] }
+  }) as PiSessionEntry;
+
+describe('runLogGateDecisions', () => {
+  let store: MemoryRunLogStore;
+
+  beforeEach(() => {
+    store = new MemoryRunLogStore();
+  });
+
+  const recorder = () => runLogGateDecisions({ store, runId: 'run-1' });
+
+  it('records a policy denial where a projector finds it', async () => {
+    const decisions = recorder();
+    decisions.onDecision(decision());
+    await decisions.flush();
+
+    const projected = projectToolState(await readAll(store, 'run-1'));
+    expect(projected.gateDecisions['call-1']).toEqual(
+      expect.objectContaining({
+        toolCallId: 'call-1',
+        toolName: 'bash',
+        verdict: 'deny',
+        decision: 'deny',
+        reason: 'destructive commands are not permitted',
+        decidedAt: '2024-01-01T00:00:00.000Z'
+      })
+    );
+  });
+
+  it('records the human who settled an asked-for call', async () => {
+    const decisions = recorder();
+    decisions.onDecision(
+      decision({
+        verdict: { decision: 'ask' },
+        decision: 'allow',
+        reason: 'looks fine',
+        actorId: 'actor-7'
+      })
+    );
+    await decisions.flush();
+
+    expect(projectToolState(await readAll(store, 'run-1')).gateDecisions['call-1']).toEqual(
+      expect.objectContaining({ verdict: 'ask', decision: 'allow', actorId: 'actor-7' })
+    );
+  });
+
+  it('hangs the entry off the log\'s leaf so the session stays one tree', async () => {
+    await store.append('run-1', [message('entry-1', 'thinking'), message('entry-2', 'calling bash')]);
+    const decisions = recorder();
+    decisions.onDecision(decision());
+    await decisions.flush();
+
+    const records = await readAll(store, 'run-1');
+    const entry = records[records.length - 1].entry;
+    expect(entry).toEqual(expect.objectContaining({ id: gateDecisionEntryId('call-1'), parentId: 'entry-2' }));
+  });
+
+  it('writes one entry when the same call is decided twice', async () => {
+    const decisions = recorder();
+    decisions.onDecision(decision());
+    decisions.onDecision(decision({ reason: 're-evaluated after a retry' }));
+    await decisions.flush();
+
+    // Deterministic ids plus the store's idempotency: a retried execution must
+    // not double-log its refusals.
+    expect(await readAll(store, 'run-1')).toHaveLength(1);
+  });
+
+  it('keeps the log in decision order', async () => {
+    const decisions = recorder();
+    decisions.onDecision(decision());
+    decisions.onDecision(decision({ toolCallId: 'call-2', toolName: 'write_file' }));
+    await decisions.flush();
+
+    const records = await readAll(store, 'run-1');
+    expect(records.map((record) => record.entry.id)).toEqual([
+      gateDecisionEntryId('call-1'),
+      gateDecisionEntryId('call-2')
+    ]);
+  });
+
+  it('throws the append failure out of flush instead of losing it', async () => {
+    const failing = runLogGateDecisions({
+      store: {
+        append: () => Promise.reject(new Error('the log is unreachable')),
+        read: store.read.bind(store)
+      } as unknown as MemoryRunLogStore,
+      runId: 'run-1'
+    });
+    failing.onDecision(decision());
+
+    await expect(failing.flush()).rejects.toThrow('the log is unreachable');
+  });
+});