From 8653ee58e4c763d78335c34e4e46df77aed7b111 Mon Sep 17 00:00:00 2001 From: Joncallim <64296013+Joncallim@users.noreply.github.com> Date: Thu, 6 Aug 2026 02:06:39 +0800 Subject: [PATCH 1/3] feat: add deterministic operation catalog foundation --- .github/workflows/web-ci.yml | 46 +- .../0011-deterministic-operation-catalog.md | 127 ++++ .../ci/prove-installer-managed-migrations.sh | 4 +- ...local-projection-overlimit-archive.test.ts | 5 +- web/__tests__/operation-adapters.test.ts | 43 ++ web/__tests__/operation-catalog.test.ts | 42 ++ web/__tests__/operation-context.test.ts | 186 ++++++ web/__tests__/operation-executor.test.ts | 378 +++++++++++ web/__tests__/operation-ledger-schema.test.ts | 21 + .../operation-ledger.postgres.test.ts | 259 ++++++++ .../operation-production-adapters.test.ts | 168 +++++ web/__tests__/repository-evidence.test.ts | 27 +- web/db/migrations/0030_operation_runs.sql | 190 ++++++ web/db/migrations/meta/_journal.json | 7 + web/db/schema.ts | 98 +++ web/lib/operations/catalog.ts | 124 ++++ web/lib/operations/contracts.ts | 177 +++++ web/lib/operations/policy.ts | 37 ++ ...prove-installer-legacy-migration-repair.sh | 4 +- .../migration-0027-expansion-assertions.sql | 6 +- .../operations/adapters/repository-read.ts | 78 +++ web/worker/operations/context.ts | 272 ++++++++ web/worker/operations/executor.ts | 624 ++++++++++++++++++ web/worker/operations/ledger.ts | 212 ++++++ web/worker/operations/production-adapters.ts | 112 ++++ web/worker/repository-evidence.ts | 12 +- web/worker/work-package-handoff.ts | 4 +- 27 files changed, 3233 insertions(+), 30 deletions(-) create mode 100644 docs/adr/0011-deterministic-operation-catalog.md create mode 100644 web/__tests__/operation-adapters.test.ts create mode 100644 web/__tests__/operation-catalog.test.ts create mode 100644 web/__tests__/operation-context.test.ts create mode 100644 web/__tests__/operation-executor.test.ts create mode 100644 web/__tests__/operation-ledger-schema.test.ts create mode 100644 web/__tests__/operation-ledger.postgres.test.ts create mode 100644 web/__tests__/operation-production-adapters.test.ts create mode 100644 web/db/migrations/0030_operation_runs.sql create mode 100644 web/lib/operations/catalog.ts create mode 100644 web/lib/operations/contracts.ts create mode 100644 web/lib/operations/policy.ts create mode 100644 web/worker/operations/adapters/repository-read.ts create mode 100644 web/worker/operations/context.ts create mode 100644 web/worker/operations/executor.ts create mode 100644 web/worker/operations/ledger.ts create mode 100644 web/worker/operations/production-adapters.ts diff --git a/.github/workflows/web-ci.yml b/.github/workflows/web-ci.yml index bb85300a..8ba26f4d 100644 --- a/.github/workflows/web-ci.yml +++ b/.github/workflows/web-ci.yml @@ -268,8 +268,11 @@ jobs: ] LOOP EXECUTE format('GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE public.%I TO forge_app_test', table_name); END LOOP; - REVOKE ALL ON TABLE public.execution_outcomes FROM forge_app_test; - GRANT SELECT, INSERT, UPDATE ON TABLE public.execution_outcomes TO forge_app_test; + REVOKE ALL ON TABLE public.execution_outcomes, public.operation_runs, + public.operation_run_events FROM forge_app_test; + GRANT SELECT, INSERT, UPDATE ON TABLE public.execution_outcomes, + public.operation_runs TO forge_app_test; + GRANT SELECT, INSERT ON TABLE public.operation_run_events TO forge_app_test; GRANT USAGE, SELECT ON SEQUENCE public.task_logs_sequence_seq TO forge_app_test; END; $grant_s4_application_acl$; @@ -509,7 +512,7 @@ jobs: 'app_settings', 'task_questions' ]; operation_ledger_tables constant text[] := ARRAY[ - 'execution_outcomes' + 'execution_outcomes', 'operation_runs', 'operation_run_events' ]; protected_tables constant text[] := ARRAY[ 'forge_release_signer_keys', 'forge_release_signer_key_lifecycle_audits', @@ -541,22 +544,37 @@ jobs: IF EXISTS ( SELECT 1 FROM pg_catalog.pg_tables WHERE schemaname = 'public' - AND tablename <> ALL (ordinary_tables || operation_ledger_tables || protected_tables || projection_tables) + AND tablename <> ALL ( + ordinary_tables || operation_ledger_tables || protected_tables || projection_tables + ) ) THEN RAISE EXCEPTION 'A public table is missing from the closed application ACL inventory'; END IF; FOREACH table_name IN ARRAY ordinary_tables LOOP EXECUTE format('GRANT SELECT, INSERT, UPDATE, DELETE ON TABLE public.%I TO forge_app_test', table_name); END LOOP; - REVOKE ALL ON TABLE public.execution_outcomes FROM forge_app_test; - GRANT SELECT, INSERT, UPDATE ON TABLE public.execution_outcomes TO forge_app_test; - FOREACH table_privilege IN ARRAY ARRAY['DELETE', 'TRUNCATE', 'REFERENCES', 'TRIGGER'] LOOP - IF has_table_privilege( - 'forge_app_test', 'public.execution_outcomes', table_privilege - ) THEN - RAISE EXCEPTION 'ordinary app has unexpected % on operation ledger table public.execution_outcomes', - table_privilege; - END IF; + FOREACH table_name IN ARRAY operation_ledger_tables LOOP + EXECUTE format('REVOKE ALL ON TABLE public.%I FROM forge_app_test', table_name); + END LOOP; + GRANT SELECT, INSERT, UPDATE ON TABLE public.execution_outcomes, + public.operation_runs TO forge_app_test; + GRANT SELECT, INSERT ON TABLE public.operation_run_events TO forge_app_test; + FOREACH table_name IN ARRAY operation_ledger_tables LOOP + FOREACH table_privilege IN ARRAY ARRAY[ + 'SELECT', 'INSERT', 'UPDATE', 'DELETE', 'TRUNCATE', 'REFERENCES', 'TRIGGER' + ] LOOP + IF has_table_privilege( + 'forge_app_test', format('public.%I', table_name), table_privilege + ) IS DISTINCT FROM ( + (table_name IN ('execution_outcomes', 'operation_runs') + AND table_privilege IN ('SELECT', 'INSERT', 'UPDATE')) + OR (table_name = 'operation_run_events' + AND table_privilege IN ('SELECT', 'INSERT')) + ) THEN + RAISE EXCEPTION 'ordinary app has unexpected % on operation ledger table public.%', + table_privilege, table_name; + END IF; + END LOOP; END LOOP; FOREACH table_name IN ARRAY protected_tables LOOP EXECUTE format('REVOKE ALL ON TABLE public.%I FROM forge_app_test', table_name); @@ -709,6 +727,8 @@ jobs: run: npm run test:unit:zero-skip env: FORGE_EPIC_172_REQUIRE_POSTGRES_TEST: '1' + FORGE_OPERATION_LEDGER_REQUIRE_POSTGRES_TEST: '1' + FORGE_OPERATION_LEDGER_POSTGRES_ADMIN_TEST_URL: postgresql://forge_e2e:forge@localhost:5432/forge_epic_172_ci_test FORGE_EPIC_172_TEST_APP_DATABASE_URL: postgresql://forge_app_test:forge_app_test@localhost:5432/forge_epic_172_ci_test FORGE_EPIC_172_TEST_WRITER_DATABASE_URL: postgresql://forge_release_evidence_writer:forge_writer_test@localhost:5432/forge_epic_172_ci_test FORGE_EPIC_172_TEST_TRANSITION_DATABASE_URL: postgresql://forge_release_transition:forge_transition_test@localhost:5432/forge_epic_172_ci_test diff --git a/docs/adr/0011-deterministic-operation-catalog.md b/docs/adr/0011-deterministic-operation-catalog.md new file mode 100644 index 00000000..3eb81590 --- /dev/null +++ b/docs/adr/0011-deterministic-operation-catalog.md @@ -0,0 +1,127 @@ +# ADR 0011: Deterministic operation catalog + +## Status + +Accepted. + +## Decision + +Forge v1 exposes a small, code-owned catalog of typed operations. An agent may +select an operation id and version, provide the exact declared input object, +state an informational reason, and name the required capability. It may not +provide a working directory, path, command, argument list, server name, or tool +name. The reason is fingerprinted for audit but cannot change execution or +idempotency. + +The initial catalog contains three read-only operations: Git status, Git diff +summary, and current Git branch. Their adapters have +fixed actions and receive project and task scope only from trusted Forge +context. They do not use a shell, write repository files, materialize generated +files, mutate GitHub or MCP state, or trigger retries. + +The production composition entry point is `executeTrustedOperation` in +`web/worker/operations/context.ts`. It joins the task to its project in +PostgreSQL, reads the current project path, root-binding revision, and project +update revision, and validates and canonicalizes an existing root through the +same project-path boundary used by normal Forge execution. Callers cannot +provide a project id or repository root. Direct construction of +`TrustedOperationContext` is reserved for focused tests and already-trusted +internal composition. + +The caller also cannot supply capabilities, ceilings, or a policy version. +Forge derives them from the current approved or running task, the linked work +package, current project revisions, and the existing effective filesystem-grant +authority. Repository reads require an approved effective +`filesystem.project.read` project grant in `always_allow` mode. Forge rejects +`allow_once` grants because this executor does not atomically consume them. +Immediately before Git starts, Forge reloads that authority and re-canonicalizes +the current project root; any task, package, grant, revision, or root change +fails closed. + +The wrapper verifies every supplied work-package, agent-run, and task-attempt +link against the authoritative task. It always composes repository reads from +Forge's bounded command runner and command-audit writer. Model output and operation request fields cannot +replace these production adapters. Each successful repository operation must +carry its command-audit UUID as evidence; exit code zero alone is insufficient. +If Git fails or cancellation wins after the command starts, the failed phase and +canonical outcome retain that audit UUID without copying raw command output into +the operation ledger. + +The diff-summary operation uses the exact argument list +`git diff --no-ext-diff --no-textconv --stat --`. The bounded command runner +rejects weaker variants, so repository attributes cannot select an external +diff helper or text-conversion program. + +Every request is checked in this order: request schema, exact inputs, catalog +version, trusted scope, existing policy ceilings, preflight, fixed adapter, and +deterministic output verification. A successful adapter call is not enough to +complete a run; verification must pass. Unknown operations, changed versions, +missing roots, denied capabilities, timeouts, malformed evidence references, +and invalid output fail closed. + +The `operation_runs` ledger stores the exact definition version, definition and +scope digests, request/input/reason fingerprints, policy decision, status, and +the linked canonical execution outcome from ADR 0010. Raw model inputs and +reasons are not stored there. Phase events are append-only and have fixed +sequence numbers from request validation through outcome. Starting a run uses a +unique task/idempotency key; final outcome creation, the outcome phase event, +and run terminalization commit in one database transaction. + +The terminal run stores a digest of the normalized canonical outcome. Replay +recomputes the digest from the linked outcome row and fails closed if that row +changed. The database permits only the explicit phase graph: validation, +policy, preflight, execution, verification, then outcome. Failed policy, +preflight, or execution phases may move directly to outcome; successful phases +cannot skip their next check. + +The scope digest binds the canonical project root, root-binding revision, +project update revision, bounded policy version, normalized capability set, and +the repository-read ceiling. A replay with the same idempotency key fails closed +if any of those inputs changed. Reordering or repeating the same capabilities +does not create a false change. + +Every adapter receives an `AbortSignal` and deadline. A timeout aborts the +signal, and Forge waits for the fixed adapter and its audit work to settle before +recording a terminal timeout. An injected adapter that ignores cancellation is +left as an incomplete recovery-required run rather than being terminalized +while work may still continue. + +A replay that finds a nonterminal `running` row also fails explicitly as +recovery-required. The incomplete row remains audit evidence; after inspection, +an operator or recovery workflow uses a new attempt key. V1 does not guess that +a stale read completed and does not mutate incomplete history. + +## Adding or changing an operation + +Add a versioned definition to `web/lib/operations/catalog.ts`, add only a fixed +adapter kind to the closed TypeScript union and executor switch, and add tests +for exact inputs, policy denial, timeout, output verification, and idempotency. +A new path, command, permission, risk, scope, executor, or verification rule is +a new version. Reviewers must confirm that all dynamic values are validated and +that existing repository, MCP, security, and human-approval ceilings remain +stricter. Project-local and model-created registrations are not supported in v1. + +To retire an operation, add its replacement first, mark the old definition +deprecated, and keep its historical version readable. Do not edit historical +ledger rows or reuse a version number. + +## Auditing + +Operators audit `operation_runs` for identity, fingerprints, policy, and the +canonical outcome link, then read `operation_run_events` in sequence order. +Evidence references are UUIDs that point to existing Forge evidence records; +technical output stays in those bounded records. A missing outcome, incomplete +phase history, digest mismatch, or invalid evidence reference is unavailable or +failed evidence, never implied success. + +## Consequences + +This foundation deliberately provides narrow read automation, not general +command authority. Write operations, rollback actions, independent workforce +verification, project-local definitions, and earned-autonomy promotion require +later reviewed versions and integrations. + +This PR establishes the production-safe composition but does not yet connect a +normal agent/model task path to `executeTrustedOperation`. That integration is +a required follow-on before issue 201 can be considered fully closed. MCP health +is also deferred until its dependency chain supports real cancellation. diff --git a/scripts/ci/prove-installer-managed-migrations.sh b/scripts/ci/prove-installer-managed-migrations.sh index a3b198f3..63827917 100755 --- a/scripts/ci/prove-installer-managed-migrations.sh +++ b/scripts/ci/prove-installer-managed-migrations.sh @@ -49,8 +49,8 @@ assert_latest_and_clean() { PGPASSWORD="$FORGE_INSTALLER_MANAGED_ADMIN_PASSWORD" PGHOST="$FORGE_INSTALLER_MANAGED_ADMIN_HOST" PGUSER="$FORGE_INSTALLER_MANAGED_ADMIN_USER" PGDATABASE="$database_name" psql --set ON_ERROR_STOP=1 <<'SQL' DO $proof$ BEGIN - IF (SELECT count(*) FROM drizzle.__drizzle_migrations) <> 30 - OR (SELECT max(created_at) FROM drizzle.__drizzle_migrations) <> 1785820800000 THEN + IF (SELECT count(*) FROM drizzle.__drizzle_migrations) <> 31 + OR (SELECT max(created_at) FROM drizzle.__drizzle_migrations) <> 1785993600000 THEN RAISE EXCEPTION 'Managed installer did not apply the exact latest migration ledger'; END IF; IF pg_catalog.to_regclass('public.forge_epic_172_s3_release_state') IS NULL THEN diff --git a/web/__tests__/local-projection-overlimit-archive.test.ts b/web/__tests__/local-projection-overlimit-archive.test.ts index b4283a20..45c1c181 100644 --- a/web/__tests__/local-projection-overlimit-archive.test.ts +++ b/web/__tests__/local-projection-overlimit-archive.test.ts @@ -378,10 +378,11 @@ describe('local-projection over-limit operator commands', () => { 'utf8', ) for (const evidence of [ - 'count(*) FROM drizzle.__drizzle_migrations) <> 30', - 'count(DISTINCT created_at) FROM drizzle.__drizzle_migrations) <> 30', + 'count(*) FROM drizzle.__drizzle_migrations) <> 31', + 'count(DISTINCT created_at) FROM drizzle.__drizzle_migrations) <> 31', 'created_at = 1784270400000', 'created_at = 1784274000000', + 'max(created_at) FROM drizzle.__drizzle_migrations) <> 1785993600000', "role.rolname = 'forge_local_projection_archiver'", 'role.rolpassword IS NULL', 'pg_catalog.pg_db_role_setting', diff --git a/web/__tests__/operation-adapters.test.ts b/web/__tests__/operation-adapters.test.ts new file mode 100644 index 00000000..4c8ace3e --- /dev/null +++ b/web/__tests__/operation-adapters.test.ts @@ -0,0 +1,43 @@ +import { describe, expect, it, vi } from 'vitest' + +import { + createRepositoryBranchReadAdapter, + createRepositoryDiffSummaryAdapter, + createRepositoryStatusReadAdapter, +} from '@/worker/operations/adapters/repository-read' + +describe('fixed repository operation adapters', () => { + it('constructs fixed argv without a shell or model-controlled path and preserves the trusted root', async () => { + const runCommand = vi.fn(async () => ({ exitCode: 0, outputSummary: 'bounded output' })) + const status = createRepositoryStatusReadAdapter({ runCommand }) + const diff = createRepositoryDiffSummaryAdapter({ runCommand }) + const branch = createRepositoryBranchReadAdapter({ runCommand }) + + await expect(status({ projectRoot: '/trusted/project' })).resolves.toMatchObject({ + output: { exitCode: 0, summary: 'bounded output' }, + }) + await expect(diff({ projectRoot: '/trusted/project' })).resolves.toMatchObject({ + output: { exitCode: 0, summary: 'bounded output' }, + }) + await expect(branch({ projectRoot: '/trusted/project' })).resolves.toMatchObject({ + output: { exitCode: 0, summary: 'bounded output' }, + }) + + expect(runCommand.mock.calls).toEqual([ + [{ cwd: '/trusted/project', command: 'git', argv: ['status', '--short'] }], + [{ cwd: '/trusted/project', command: 'git', argv: ['diff', '--no-ext-diff', '--no-textconv', '--stat', '--'] }], + [{ cwd: '/trusted/project', command: 'git', argv: ['branch', '--show-current'] }], + ]) + }) + + it.each(['relative/project', '../outside', 'trusted/project\0outside'])( + 'rejects an unsafe trusted root before invoking the command boundary: %s', + async (projectRoot) => { + const runCommand = vi.fn(async () => ({ exitCode: 0, outputSummary: '' })) + const status = createRepositoryStatusReadAdapter({ runCommand }) + + await expect(status({ projectRoot })).rejects.toThrow('absolute filesystem path') + expect(runCommand).not.toHaveBeenCalled() + }, + ) +}) diff --git a/web/__tests__/operation-catalog.test.ts b/web/__tests__/operation-catalog.test.ts new file mode 100644 index 00000000..2d8224b3 --- /dev/null +++ b/web/__tests__/operation-catalog.test.ts @@ -0,0 +1,42 @@ +import { describe, expect, it } from 'vitest' + +import { + BUILT_IN_OPERATIONS, + createOperationCatalog, + resolveOperationDefinition, + validateOperationInputs, +} from '@/lib/operations/catalog' +import { OperationContractError, parseOperationRequest } from '@/lib/operations/contracts' + +function request(overrides: Record = {}) { + return { + schemaVersion: 1, + operationId: 'repository.status.read', + operationVersion: 1, + inputs: {}, + reason: 'Inspect repository state.', + requestedCapability: 'filesystem.project.read', + ...overrides, + } +} + +describe('operation catalog v1', () => { + it('requires schemaVersion 1 and exact request keys', () => { + expect(() => parseOperationRequest(request({ schemaVersion: 2 }))).toThrow(OperationContractError) + expect(() => parseOperationRequest(request({ cwd: '/tmp/escape' }))).toThrow('exactly the v1 request keys') + }) + + it('preserves an exact empty input object and rejects command or path injection', () => { + const parsed = parseOperationRequest(request()) + expect(parsed.inputs).toEqual({}) + const definition = resolveOperationDefinition(parsed) + expect(() => validateOperationInputs(definition, parsed.inputs)).not.toThrow() + expect(() => validateOperationInputs(definition, { argv: ['sh', '-c', 'rm -rf /'] })).toThrow(OperationContractError) + expect(() => validateOperationInputs(definition, { projectPath: '../../outside' })).toThrow(OperationContractError) + }) + + it('rejects duplicate ids and unsupported versions', () => { + expect(() => createOperationCatalog([BUILT_IN_OPERATIONS[0], BUILT_IN_OPERATIONS[0]])).toThrow('Duplicate') + expect(() => resolveOperationDefinition({ operationId: 'repository.status.read', operationVersion: 2 })).toThrow('version') + }) +}) diff --git a/web/__tests__/operation-context.test.ts b/web/__tests__/operation-context.test.ts new file mode 100644 index 00000000..7a33c65f --- /dev/null +++ b/web/__tests__/operation-context.test.ts @@ -0,0 +1,186 @@ +import { describe, expect, it, vi } from 'vitest' + +import { loadTrustedOperationContextForTest } from '@/worker/operations/context' + +const taskId = '11111111-1111-4111-8111-111111111111' +const projectId = '22222222-2222-4222-8222-222222222222' +const workPackageId = '33333333-3333-4333-8333-333333333333' + +function joined(overrides: Record = {}) { + return { + taskId, + taskStatus: 'approved', + taskUpdatedAt: new Date('2026-08-06T00:00:00.000Z'), + projectId, + localPath: '/uncanonical/project', + grantDecisionRevision: BigInt(1), + rootBindingRevision: BigInt(12), + projectUpdatedAt: new Date('2026-08-06T00:00:00.000Z'), + projectMcpConfig: { profile: 'default', requiredMcps: [], overrides: {} }, + ...overrides, + } +} + +function approvedAllowOnceMetadata() { + return { + mcpGrantPhases: { + effective: { + schemaVersion: 2, + phase: 'effective', + source: 'explicit-grant-approval', + grantMode: 'allow_once', + runtimeIssued: false, + runtimeEnforcement: 'bounded_context_packet', + status: 'approved', + grantDecisionRevision: '1', + rootBindingRevision: '12', + grants: [{ mcpId: 'filesystem', status: 'approved', capabilities: ['filesystem.project.read'] }], + }, + }, + } +} + +function approvedAlwaysAllowMetadata() { + return { + mcpGrantPhases: { + effective: { + schemaVersion: 2, + phase: 'effective', + source: 'project-filesystem-approval', + runtimeEnforcement: 'bounded_context_packet', + status: 'approved', + grantDecisionRevision: '1', + rootBindingRevision: '12', + grants: [{ mcpId: 'filesystem', status: 'approved', capabilities: ['filesystem.project.read'] }], + }, + }, + } +} + +function approvedProjectDecision() { + return { + schemaVersion: 2, + decisionId: 'decision-1', + projectId, + decision: 'approved', + capabilities: ['filesystem.project.read'], + grantDecisionRevision: '1', + rootBindingRevision: '12', + decisionFingerprint: `sha256:${'a'.repeat(64)}`, + decisionGeneration: '1', + decidedAt: '2026-08-06T00:00:00.000Z', + decidedBy: 'user-1', + reason: 'Approved project read authority.', + revocationReason: null, + } +} + +describe('trusted operation context loader', () => { + it('derives repository policy from the authoritative active task and linked package grant', async () => { + const canonicalizeProjectRoot = vi.fn(async () => '/workspace/projects/forge') + const context = await loadTrustedOperationContextForTest({ + taskId, + attemptKey: 'attempt-1', + workPackageId, + }, { + loadTaskProject: vi.fn(async () => joined()), + canonicalizeProjectRoot, + loadLinkedIds: vi.fn(async () => ({ + workPackageId, + workPackageMetadata: approvedAlwaysAllowMetadata(), + workPackageUpdatedAt: new Date('2026-08-06T00:00:01.000Z'), + agentRunId: null, + taskAttemptId: null, + })), + loadProjectFilesystemDecision: vi.fn(async () => approvedProjectDecision()), + }) + + expect(canonicalizeProjectRoot).toHaveBeenCalledWith({ id: projectId, localPath: '/uncanonical/project' }) + expect(context).toMatchObject({ + taskId, + projectId, + projectRoot: '/workspace/projects/forge', + rootBindingRevision: '12', + projectRevision: '2026-08-06T00:00:00.000Z', + policy: { + projectId, + scopedProjectId: projectId, + allowedCapabilities: ['filesystem.project.read'], + policyCeilings: { repository_read: true }, + }, + }) + expect(context.policy.policyVersion).toMatch(/^[0-9a-f]{64}$/) + }) + + it('does not grant repository operations from an unconsumed allow-once grant', async () => { + const context = await loadTrustedOperationContextForTest({ taskId, attemptKey: 'attempt-1', workPackageId }, { + loadTaskProject: vi.fn(async () => joined()), + canonicalizeProjectRoot: vi.fn(async () => '/workspace/projects/forge'), + loadLinkedIds: vi.fn(async () => ({ + workPackageId, + workPackageMetadata: approvedAllowOnceMetadata(), + workPackageUpdatedAt: new Date('2026-08-06T00:00:01.000Z'), + agentRunId: null, + taskAttemptId: null, + })), + }) + expect(context.policy).toMatchObject({ + allowedCapabilities: [], + policyCeilings: { repository_read: false }, + }) + }) + + it('does not grant repository reads without an authoritative linked package', async () => { + const canonicalizeProjectRoot = vi.fn() + const context = await loadTrustedOperationContextForTest({ taskId, attemptKey: 'attempt-1' }, { + loadTaskProject: vi.fn(async () => joined({ localPath: null, rootBindingRevision: BigInt(0) })), + canonicalizeProjectRoot, + }) + expect(context.projectRoot).toBeNull() + expect(context.policy).toMatchObject({ allowedCapabilities: [], policyCeilings: { repository_read: false } }) + expect(canonicalizeProjectRoot).not.toHaveBeenCalled() + }) + + it('rejects inactive tasks, mismatched joins, and unrelated run links', async () => { + const canonicalizeProjectRoot = vi.fn() + await expect(loadTrustedOperationContextForTest({ taskId, attemptKey: 'attempt-1' }, { + loadTaskProject: vi.fn(async () => joined({ taskStatus: 'completed' })), + canonicalizeProjectRoot, + })).rejects.toThrow('approved or running') + + await expect(loadTrustedOperationContextForTest({ taskId, attemptKey: 'attempt-1' }, { + loadTaskProject: vi.fn(async () => joined({ taskId: '44444444-4444-4444-8444-444444444444' })), + canonicalizeProjectRoot, + })).rejects.toThrow('could not be resolved') + + await expect(loadTrustedOperationContextForTest({ taskId, attemptKey: 'attempt-1', workPackageId }, { + loadTaskProject: vi.fn(async () => joined({ localPath: null })), + canonicalizeProjectRoot, + loadLinkedIds: vi.fn(async () => null), + })).rejects.toThrow('do not belong') + }) + + it('changes authoritative policyVersion when task or package revision changes', async () => { + const load = (taskUpdatedAt: string, packageUpdatedAt: string) => loadTrustedOperationContextForTest({ + taskId, + attemptKey: 'attempt-1', + workPackageId, + }, { + loadTaskProject: vi.fn(async () => joined({ taskUpdatedAt: new Date(taskUpdatedAt) })), + canonicalizeProjectRoot: vi.fn(async () => '/workspace/projects/forge'), + loadLinkedIds: vi.fn(async () => ({ + workPackageId, + workPackageMetadata: approvedAlwaysAllowMetadata(), + workPackageUpdatedAt: new Date(packageUpdatedAt), + agentRunId: null, + taskAttemptId: null, + })), + loadProjectFilesystemDecision: vi.fn(async () => approvedProjectDecision()), + }) + const baseline = await load('2026-08-06T00:00:00.000Z', '2026-08-06T00:00:01.000Z') + const taskChanged = await load('2026-08-06T00:00:02.000Z', '2026-08-06T00:00:01.000Z') + const packageChanged = await load('2026-08-06T00:00:00.000Z', '2026-08-06T00:00:03.000Z') + expect(taskChanged.policy.policyVersion).not.toBe(baseline.policy.policyVersion) + expect(packageChanged.policy.policyVersion).not.toBe(baseline.policy.policyVersion) + }) +}) diff --git a/web/__tests__/operation-executor.test.ts b/web/__tests__/operation-executor.test.ts new file mode 100644 index 00000000..6dd4e5d4 --- /dev/null +++ b/web/__tests__/operation-executor.test.ts @@ -0,0 +1,378 @@ +import { describe, expect, it, vi } from 'vitest' + +import { BUILT_IN_OPERATIONS, createOperationCatalog } from '@/lib/operations/catalog' +import type { OperationExecutionResult } from '@/lib/operations/contracts' +import { createRepositoryStatusReadAdapter } from '@/worker/operations/adapters/repository-read' +import { + OperationAdapterExecutionError, + executeOperation, + type FixedOperationAdapters, + type OperationLedger, + type OperationRunEventWrite, + type OperationRunFinalization, + type OperationRunStart, + type TrustedOperationContext, +} from '@/worker/operations/executor' + +const taskId = '11111111-1111-4111-8111-111111111111' +const projectId = '22222222-2222-4222-8222-222222222222' +const runId = '33333333-3333-4333-8333-333333333333' +const outcomeId = '44444444-4444-4444-8444-444444444444' +const evidenceId = '55555555-5555-4555-8555-555555555555' + +class FakeLedger implements OperationLedger { + start: OperationRunStart | null = null + events: OperationRunEventWrite[] = [] + finalization: OperationRunFinalization | null = null + replay: OperationExecutionResult | null = null + + async begin(input: OperationRunStart) { + this.start = input + if (this.replay) { + return { + kind: 'replayed' as const, + requestFingerprint: input.requestFingerprint, + definitionDigest: input.definitionDigest, + scopeFingerprint: input.scopeFingerprint, + result: this.replay, + } + } + return { kind: 'started' as const, runId } + } + + async appendEvent(event: OperationRunEventWrite) { + this.events.push(event) + } + + async finalize(input: OperationRunFinalization) { + this.finalization = input + this.events.push(input.outcomeEvent) + return { executionOutcomeId: outcomeId } + } +} + +function context(overrides: Partial = {}): TrustedOperationContext { + return { + taskId, + projectId, + attemptKey: 'attempt-1', + projectRoot: '/trusted/project', + rootBindingRevision: '7', + projectRevision: '2026-08-06T00:00:00.000Z', + policy: { + projectId, + scopedProjectId: projectId, + policyVersion: 'policy-v1', + allowedCapabilities: ['filesystem.project.read'], + policyCeilings: { repository_read: true }, + }, + ...overrides, + } +} + +function request(overrides: Record = {}) { + return { + schemaVersion: 1, + operationId: 'repository.status.read', + operationVersion: 1, + inputs: {}, + reason: 'Read status.', + requestedCapability: 'filesystem.project.read', + ...overrides, + } +} + +function adapters(overrides: Partial = {}): FixedOperationAdapters { + return { + repositoryStatusRead: vi.fn(async () => ({ output: { exitCode: 0, summary: 'M file.ts' }, evidenceRefs: [evidenceId] })), + repositoryDiffSummary: vi.fn(async () => ({ output: { exitCode: 0, summary: '1 file changed' }, evidenceRefs: [evidenceId] })), + repositoryBranchRead: vi.fn(async () => ({ output: { exitCode: 0, summary: 'main' }, evidenceRefs: [evidenceId] })), + ...overrides, + } +} + +describe('deterministic operation executor', () => { + it('records fixed ordered phases and links a verified canonical outcome', async () => { + const ledger = new FakeLedger() + const result = await executeOperation({ request: request(), context: context(), dependencies: { ledger, adapters: adapters() } }) + + expect(result).toMatchObject({ status: 'completed', verificationStatus: 'passed', replayed: false }) + expect(result.definitionDigest).toMatch(/^[0-9a-f]{64}$/) + expect(result.scopeFingerprint).toMatch(/^[0-9a-f]{64}$/) + expect(ledger.start?.inputsFingerprint).toMatch(/^[0-9a-f]{64}$/) + expect(ledger.events.map(({ sequence, phase }) => [sequence, phase])).toEqual([ + [0, 'request_validation'], [1, 'policy'], [2, 'preflight'], + [3, 'execution'], [4, 'verification'], [5, 'outcome'], + ]) + expect(ledger.finalization?.outcome).toMatchObject({ result: 'completed', stopReasonCode: null }) + }) + + it('fails invalid input before the ledger or adapter sees it', async () => { + const ledger = new FakeLedger() + const fixed = adapters() + await expect(executeOperation({ + request: request({ inputs: { cwd: '/tmp', argv: ['sh'] } }), + context: context(), + dependencies: { ledger, adapters: fixed }, + })).rejects.toThrow('Inputs do not match') + expect(ledger.start).toBeNull() + expect(fixed.repositoryStatusRead).not.toHaveBeenCalled() + }) + + it('blocks on the trusted policy ceiling before adapter execution', async () => { + const ledger = new FakeLedger() + const fixed = adapters() + const trusted = context({ + policy: { ...context().policy, policyCeilings: { repository_read: false } }, + }) + const result = await executeOperation({ request: request(), context: trusted, dependencies: { ledger, adapters: fixed } }) + expect(result.status).toBe('blocked') + expect(result.outcome?.stopReasonCode).toBe('policy_blocked') + expect(fixed.repositoryStatusRead).not.toHaveBeenCalled() + expect(ledger.events.map((event) => event.phase)).toEqual(['request_validation', 'policy', 'outcome']) + }) + + it('does not treat adapter transport success as verification success', async () => { + const ledger = new FakeLedger() + const fixed = adapters({ + repositoryStatusRead: vi.fn(async () => ({ output: { exitCode: 0, summary: 'transport only' }, evidenceRefs: [] })), + }) + const result = await executeOperation({ request: request(), context: context(), dependencies: { ledger, adapters: fixed } }) + expect(result).toMatchObject({ status: 'failed', verificationStatus: 'failed' }) + expect(result.outcome?.stopReasonCode).toBe('validation_failed') + }) + + it('fails closed on timeout', async () => { + const ledger = new FakeLedger() + const timeoutDefinition = { ...BUILT_IN_OPERATIONS[0], timeoutMs: 5 } + let observedSignal: AbortSignal | null = null + let adapterSettled = false + const fixed = adapters({ + repositoryStatusRead: vi.fn((_context, execution) => { + observedSignal = execution.signal + return new Promise((_resolve, reject) => { + execution.signal.addEventListener('abort', () => { + adapterSettled = true + reject(new OperationAdapterExecutionError('cancelled', [evidenceId])) + }, { once: true }) + }) + }), + }) + const result = await executeOperation({ + request: request(), + context: context(), + dependencies: { ledger, adapters: fixed, catalog: createOperationCatalog([timeoutDefinition]) }, + }) + expect(result.status).toBe('failed') + expect(result.outcome?.stopReasonCode).toBe('timeout') + expect((observedSignal as unknown as AbortSignal).aborted).toBe(true) + expect(adapterSettled).toBe(true) + expect(ledger.events.find((event) => event.phase === 'execution')?.evidenceRefs).toEqual([evidenceId]) + expect(ledger.finalization?.outcome.evidenceRefs).toEqual([evidenceId]) + }) + + it('links typed adapter-failure audit evidence without exposing raw output', async () => { + const ledger = new FakeLedger() + const result = await executeOperation({ + request: request(), + context: context(), + dependencies: { + ledger, + adapters: adapters({ + repositoryStatusRead: vi.fn(async () => { + throw new OperationAdapterExecutionError('adapter_failed', [evidenceId]) + }), + }), + }, + }) + expect(result).toMatchObject({ status: 'failed', output: null }) + expect(result.outcome).toMatchObject({ evidenceRefs: [evidenceId], stopReasonSummary: 'The deterministic operation execution failed.' }) + expect(ledger.events.find((event) => event.phase === 'execution')?.evidenceRefs).toEqual([evidenceId]) + }) + + it('leaves the run incomplete when an injected adapter ignores cancellation', async () => { + const ledger = new FakeLedger() + const timeoutDefinition = { ...BUILT_IN_OPERATIONS[0], timeoutMs: 5 } + await expect(executeOperation({ + request: request(), + context: context(), + dependencies: { + ledger, + adapters: adapters({ repositoryStatusRead: vi.fn(() => new Promise(() => {})) }), + catalog: createOperationCatalog([timeoutDefinition]), + }, + })).rejects.toThrow('requires recovery') + expect(ledger.finalization).toBeNull() + expect(ledger.events.map((event) => event.phase)).toEqual(['request_validation', 'policy', 'preflight']) + }) + + it('keeps the audit-only reason out of request idempotency', async () => { + const first = new FakeLedger() + const second = new FakeLedger() + await executeOperation({ request: request({ reason: 'First audit note.' }), context: context(), dependencies: { ledger: first, adapters: adapters() } }) + await executeOperation({ request: request({ reason: 'Different audit note.' }), context: context(), dependencies: { ledger: second, adapters: adapters() } }) + expect(first.start?.idempotencyKey).toBe(second.start?.idempotencyKey) + expect(first.start?.requestFingerprint).toBe(second.start?.requestFingerprint) + expect(first.start?.reasonFingerprint).not.toBe(second.start?.reasonFingerprint) + }) + + it('binds replay scope to normalized policy and project revisions', async () => { + const baseline = new FakeLedger() + const reordered = new FakeLedger() + const policyChanged = new FakeLedger() + const rootChanged = new FakeLedger() + const projectChanged = new FakeLedger() + await executeOperation({ request: request(), context: context(), dependencies: { ledger: baseline, adapters: adapters() } }) + await executeOperation({ + request: request(), + context: context({ + policy: { + ...context().policy, + allowedCapabilities: ['filesystem.project.read', 'filesystem.project.read'], + }, + }), + dependencies: { ledger: reordered, adapters: adapters() }, + }) + await executeOperation({ + request: request(), + context: context({ policy: { ...context().policy, policyVersion: 'policy-v2' } }), + dependencies: { ledger: policyChanged, adapters: adapters() }, + }) + await executeOperation({ + request: request(), + context: context({ rootBindingRevision: '8' }), + dependencies: { ledger: rootChanged, adapters: adapters() }, + }) + await executeOperation({ + request: request(), + context: context({ projectRevision: '2026-08-06T00:01:00.000Z' }), + dependencies: { ledger: projectChanged, adapters: adapters() }, + }) + + expect(reordered.start?.scopeFingerprint).toBe(baseline.start?.scopeFingerprint) + expect(policyChanged.start?.scopeFingerprint).not.toBe(baseline.start?.scopeFingerprint) + expect(rootChanged.start?.scopeFingerprint).not.toBe(baseline.start?.scopeFingerprint) + expect(projectChanged.start?.scopeFingerprint).not.toBe(baseline.start?.scopeFingerprint) + }) + + it('keeps idempotency stable per operation and distinct across operations', async () => { + const status = new FakeLedger() + const diff = new FakeLedger() + const branch = new FakeLedger() + await executeOperation({ request: request(), context: context(), dependencies: { ledger: status, adapters: adapters() } }) + await executeOperation({ + request: request({ operationId: 'repository.diff.summary' }), + context: context(), + dependencies: { ledger: diff, adapters: adapters() }, + }) + await executeOperation({ + request: request({ + operationId: 'repository.branch.read', + }), + context: context(), + dependencies: { ledger: branch, adapters: adapters() }, + }) + + expect(new Set([ + status.start?.idempotencyKey, + diff.start?.idempotencyKey, + branch.start?.idempotencyKey, + ])).toHaveProperty('size', 3) + expect(new Set([ + status.finalization?.attemptKey, + diff.finalization?.attemptKey, + branch.finalization?.attemptKey, + ])).toHaveProperty('size', 3) + }) + + it('records trusted-scope failure as preflight, before adapter execution', async () => { + const ledger = new FakeLedger() + const fixed = adapters() + const result = await executeOperation({ + request: request(), + context: context({ projectRoot: null }), + dependencies: { ledger, adapters: fixed }, + }) + expect(result.status).toBe('failed') + expect(result.outcome?.stopReasonCode).toBe('missing_repository_context') + expect(fixed.repositoryStatusRead).not.toHaveBeenCalled() + expect(ledger.events.map((event) => [event.phase, event.status])).toEqual([ + ['request_validation', 'passed'], + ['policy', 'passed'], + ['preflight', 'failed'], + ['outcome', 'failed'], + ]) + }) + + it('attributes a non-absolute trusted repository root to preflight', async () => { + const ledger = new FakeLedger() + const runCommand = vi.fn(async () => ({ exitCode: 0, outputSummary: '' })) + const fixed = adapters({ + repositoryStatusRead: createRepositoryStatusReadAdapter({ runCommand }), + }) + const result = await executeOperation({ + request: request(), + context: context({ projectRoot: '../outside' }), + dependencies: { ledger, adapters: fixed }, + }) + + expect(result.status).toBe('failed') + expect(runCommand).not.toHaveBeenCalled() + expect(ledger.events.map((event) => [event.phase, event.status, event.detailCode])).toEqual([ + ['request_validation', 'passed', 'request_valid'], + ['policy', 'passed', 'allowed'], + ['preflight', 'failed', 'preflight_failed'], + ['outcome', 'failed', 'missing_repository_context'], + ]) + }) + + it('rejects invalid evidence references and replays a completed idempotent run without execution', async () => { + const invalidLedger = new FakeLedger() + const invalid = adapters({ + repositoryStatusRead: vi.fn(async () => ({ output: { exitCode: 0, summary: '' }, evidenceRefs: ['not-a-uuid'] })), + }) + const failed = await executeOperation({ request: request(), context: context(), dependencies: { ledger: invalidLedger, adapters: invalid } }) + expect(failed.status).toBe('failed') + + const ledger = new FakeLedger() + ledger.replay = { + runId, + operationId: 'repository.status.read', + operationVersion: 1, + definitionDigest: 'a'.repeat(64), + scopeFingerprint: 'b'.repeat(64), + status: 'completed', + verificationStatus: 'passed', + replayed: false, + output: { shouldNotReplay: true }, + outcome: null, + } + const fixed = adapters() + const replayed = await executeOperation({ request: request(), context: context(), dependencies: { ledger, adapters: fixed } }) + expect(replayed).toMatchObject({ replayed: true, output: null }) + expect(fixed.repositoryStatusRead).not.toHaveBeenCalled() + }) + + it('does not return a nonterminal running row as a completed replay', async () => { + const ledger = new FakeLedger() + ledger.replay = { + runId, + operationId: 'repository.status.read', + operationVersion: 1, + definitionDigest: 'a'.repeat(64), + scopeFingerprint: 'b'.repeat(64), + status: 'running', + verificationStatus: 'not_started', + replayed: false, + output: null, + outcome: null, + } + const fixed = adapters() + await expect(executeOperation({ + request: request(), + context: context(), + dependencies: { ledger, adapters: fixed }, + })).rejects.toThrow('still in progress or requires recovery') + expect(fixed.repositoryStatusRead).not.toHaveBeenCalled() + }) +}) diff --git a/web/__tests__/operation-ledger-schema.test.ts b/web/__tests__/operation-ledger-schema.test.ts new file mode 100644 index 00000000..18c6d705 --- /dev/null +++ b/web/__tests__/operation-ledger-schema.test.ts @@ -0,0 +1,21 @@ +import fs from 'node:fs/promises' +import path from 'node:path' +import { describe, expect, it } from 'vitest' + +describe('operation ledger migration', () => { + it('persists immutable digests, policy, ordered events, and canonical outcome linkage', async () => { + const sql = await fs.readFile(path.join(process.cwd(), 'db/migrations/0030_operation_runs.sql'), 'utf8') + expect(sql).toContain('"definition_digest" text NOT NULL') + expect(sql).toContain('"scope_fingerprint" text NOT NULL') + expect(sql).toContain('"outcome_fingerprint" text') + expect(sql).toContain('"policy_decision" jsonb NOT NULL') + expect(sql).toContain('"execution_outcome_id" uuid') + expect(sql).toContain('operation_runs_fingerprints_check') + expect(sql).toContain('operation_run_events_phase_sequence_check') + expect(sql).toContain('operation_run_events_order_guard') + expect(sql).toContain("v_last_sequence = 1") + expect(sql).toContain("v_last_status = 'passed'") + expect(sql).toContain('outcome_fingerprint') + expect(sql).toContain('operation_run_events_append_only') + }) +}) diff --git a/web/__tests__/operation-ledger.postgres.test.ts b/web/__tests__/operation-ledger.postgres.test.ts new file mode 100644 index 00000000..c3405482 --- /dev/null +++ b/web/__tests__/operation-ledger.postgres.test.ts @@ -0,0 +1,259 @@ +import { randomUUID } from 'node:crypto' + +import postgres from 'postgres' +import { afterAll, beforeAll, describe, expect, it } from 'vitest' + +import { + operationFingerprint, + type OperationPolicyDecision, +} from '@/lib/operations/contracts' +import type { + OperationLedger, + OperationRunEventWrite, + OperationRunFinalization, + OperationRunStart, +} from '@/worker/operations/executor' + +const required = process.env.FORGE_OPERATION_LEDGER_REQUIRE_POSTGRES_TEST === '1' +const databaseUrl = process.env.DATABASE_URL?.trim() +const adminUrl = process.env.FORGE_OPERATION_LEDGER_POSTGRES_ADMIN_TEST_URL?.trim() +const enabled = required && Boolean(databaseUrl && adminUrl) + +if (required && (!databaseUrl || !adminUrl)) { + throw new Error( + 'FORGE_OPERATION_LEDGER_REQUIRE_POSTGRES_TEST=1 requires DATABASE_URL and FORGE_OPERATION_LEDGER_POSTGRES_ADMIN_TEST_URL for the disposable PostgreSQL operation-ledger proof; the mandatory suite may not skip.', + ) +} + +describe.skipIf(!enabled)('operation ledger PostgreSQL behavior', () => { + const ids = { + user: randomUUID(), + project: randomUUID(), + task: randomUUID(), + } + const idempotencyKey = operationFingerprint('postgres-proof-idempotency', randomUUID()) + const outcomeAttemptKey = `operation:${operationFingerprint('postgres-proof-outcome', randomUUID())}` + const policyDecision: OperationPolicyDecision = { + schemaVersion: 1, + allowed: true, + code: 'allowed', + capability: 'filesystem.project.read', + projectId: ids.project, + policyVersion: 'operation-ledger-postgres-v1', + } + const start: OperationRunStart = { + taskId: ids.task, + projectId: ids.project, + workPackageId: null, + agentRunId: null, + taskAttemptId: null, + definitionSchemaVersion: 1, + operationId: 'repository.status.read', + operationVersion: 1, + capability: 'filesystem.project.read', + idempotencyKey, + definitionDigest: operationFingerprint('postgres-proof-definition', { version: 1 }), + scopeFingerprint: operationFingerprint('postgres-proof-scope', { projectId: ids.project }), + requestFingerprint: operationFingerprint('postgres-proof-request', { operationId: 'repository.status.read' }), + inputsFingerprint: operationFingerprint('postgres-proof-inputs', {}), + reasonFingerprint: operationFingerprint('postgres-proof-reason', 'live ledger proof'), + policyDecision, + } + const outcome = { + schemaVersion: 1 as const, + transportStatus: 'ok' as const, + result: 'completed' as const, + stopReasonCode: null, + stopReasonSummary: null, + retryable: false, + evidenceRefs: [], + verifierRequired: false, + verificationStatus: 'not_required' as const, + } + + let sql: ReturnType + let adminSql: ReturnType + let factoryLedger: OperationLedger + let defaultLedger: OperationLedger + let closeDb: () => Promise + + function event( + runId: string, + sequence: number, + phase: OperationRunEventWrite['phase'], + status: OperationRunEventWrite['status'] = 'passed', + ): OperationRunEventWrite { + return { + runId, + sequence, + phase, + status, + detailCode: `${phase}_${status}`, + detailFingerprint: operationFingerprint(`postgres-proof-event:${phase}`, { runId, sequence, status }), + evidenceRefs: [], + } + } + + beforeAll(async () => { + sql = postgres(databaseUrl!, { max: 4, onnotice: () => {} }) + adminSql = postgres(adminUrl!, { max: 1, onnotice: () => {} }) + const ledgerModule = await import('@/worker/operations/ledger') + const databaseModule = await import('@/db') + factoryLedger = ledgerModule.createDatabaseOperationLedger() + defaultLedger = ledgerModule.databaseOperationLedger + closeDb = databaseModule.closeDb + + await sql.begin(async (tx) => { + await tx` + insert into users (id, display_name) + values (${ids.user}::uuid, 'Operation ledger PostgreSQL proof') + ` + await tx` + insert into projects (id, name, submitted_by, grant_decision_revision, root_binding_revision) + values (${ids.project}::uuid, 'Operation ledger PostgreSQL proof', ${ids.user}::uuid, 1, 1) + ` + await tx` + insert into tasks (id, project_id, submitted_by, title, prompt, status) + values ( + ${ids.task}::uuid, ${ids.project}::uuid, ${ids.user}::uuid, + 'Operation ledger PostgreSQL proof', 'Bounded disposable test fixture', 'running' + ) + ` + }) + }) + + afterAll(async () => { + await closeDb?.() + await Promise.all([ + sql?.end({ timeout: 5 }), + adminSql?.end({ timeout: 5 }), + ]) + }) + + it('enforces concurrency, ordered append-only phases, atomic finalization, replay, and outcome integrity', async () => { + const begins = await Promise.all([ + factoryLedger.begin(start), + defaultLedger.begin(start), + ]) + const started = begins.filter((result) => result.kind === 'started') + const concurrentReplay = begins.filter((result) => result.kind === 'replayed') + expect(started).toHaveLength(1) + expect(concurrentReplay).toHaveLength(1) + expect(concurrentReplay[0]).toMatchObject({ + kind: 'replayed', + result: { status: 'running', verificationStatus: 'not_started', outcome: null }, + }) + const runId = started[0]!.runId + + await expect(defaultLedger.appendEvent(event(runId, 1, 'policy'))) + .rejects.toThrow('operation run events must be appended in phase order') + + for (const [sequence, phase] of [ + [0, 'request_validation'], + [1, 'policy'], + [2, 'preflight'], + [3, 'execution'], + [4, 'verification'], + ] as const) { + await factoryLedger.appendEvent(event(runId, sequence, phase)) + } + + const finalization: OperationRunFinalization = { + runId, + taskId: ids.task, + workPackageId: null, + agentRunId: null, + taskAttemptId: null, + attemptKey: outcomeAttemptKey, + status: 'completed', + verificationStatus: 'passed', + outputFingerprint: operationFingerprint('postgres-proof-output', { exitCode: 0, summary: '' }), + policyDecision, + outcome, + outcomeEvent: event(runId, 5, 'outcome'), + } + + await expect(defaultLedger.finalize({ + ...finalization, + outcomeEvent: event(runId, 4, 'verification'), + })).rejects.toThrow('operation run events must be appended in phase order') + const [rolledBack] = await sql<{ + outcomeCount: number + status: string + executionOutcomeId: string | null + }[]>` + select + (select count(*)::int from execution_outcomes where task_id = ${ids.task}::uuid and attempt_key = ${outcomeAttemptKey}) as "outcomeCount", + status, + execution_outcome_id as "executionOutcomeId" + from operation_runs + where id = ${runId}::uuid + ` + expect(rolledBack).toEqual({ outcomeCount: 0, status: 'running', executionOutcomeId: null }) + + const finalized = await defaultLedger.finalize(finalization) + const [persisted] = await sql<{ + executionOutcomeId: string + outcomeFingerprint: string + status: string + verificationStatus: string + phases: string[] + }[]>` + select + run.execution_outcome_id as "executionOutcomeId", + run.outcome_fingerprint as "outcomeFingerprint", + run.status, + run.verification_status as "verificationStatus", + array( + select event_row.phase + from operation_run_events event_row + where event_row.operation_run_id = run.id + order by event_row.sequence + ) as phases + from operation_runs run + where run.id = ${runId}::uuid + ` + expect(persisted).toMatchObject({ + executionOutcomeId: finalized.executionOutcomeId, + status: 'completed', + verificationStatus: 'passed', + phases: ['request_validation', 'policy', 'preflight', 'execution', 'verification', 'outcome'], + }) + expect(persisted?.outcomeFingerprint).toMatch(/^[0-9a-f]{64}$/) + + const replay = await factoryLedger.begin(start) + expect(replay).toMatchObject({ + kind: 'replayed', + result: { + runId, + status: 'completed', + verificationStatus: 'passed', + outcome: { result: 'completed', stopReasonCode: null }, + }, + }) + + await expect(defaultLedger.appendEvent(event(runId, 5, 'outcome'))) + .rejects.toThrow('terminal operation runs cannot receive events') + await expect(sql`update operation_runs set status = 'failed' where id = ${runId}::uuid`) + .rejects.toThrow('terminal operation runs are immutable') + const [eventPrivileges] = await sql<{ canUpdate: boolean }[]>` + select has_table_privilege( + current_user, 'public.operation_run_events', 'UPDATE' + ) as "canUpdate" + ` + expect(eventPrivileges).toEqual({ canUpdate: false }) + await expect(adminSql` + update operation_run_events + set detail_code = 'tampered' + where operation_run_id = ${runId}::uuid and sequence = 0 + `).rejects.toThrow('operation run events are append-only') + + await sql` + update execution_outcomes + set stop_reason_summary = 'tampered after terminalization' + where id = ${finalized.executionOutcomeId}::uuid + ` + await expect(defaultLedger.begin(start)) + .rejects.toThrow('Stored canonical operation outcome does not match its immutable fingerprint') + }) +}) diff --git a/web/__tests__/operation-production-adapters.test.ts b/web/__tests__/operation-production-adapters.test.ts new file mode 100644 index 00000000..4dedc3aa --- /dev/null +++ b/web/__tests__/operation-production-adapters.test.ts @@ -0,0 +1,168 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ + assertProjectLocalPathForExecution: vi.fn(), + loadTrustedOperationContext: vi.fn(), + recordScopedCommandAudit: vi.fn(), + runScopedRepositoryCommand: vi.fn(), +})) + +const project = { + id: '22222222-2222-4222-8222-222222222222', + localPath: '/trusted/project', + rootBindingRevision: BigInt(7), + updatedAt: new Date('2026-08-06T00:00:00.000Z'), +} + +vi.mock('@/db', () => ({ + db: { + select: vi.fn(() => ({ + from: vi.fn(() => ({ + where: vi.fn(() => ({ limit: vi.fn(async () => [project]) })), + })), + })), + }, +})) +vi.mock('@/lib/projects/local-path', () => ({ + assertProjectLocalPathForExecution: mocks.assertProjectLocalPathForExecution, +})) +vi.mock('@/worker/operations/context', () => ({ + loadTrustedOperationContext: mocks.loadTrustedOperationContext, +})) +vi.mock('@/worker/repository-evidence', () => ({ + recordScopedCommandAudit: mocks.recordScopedCommandAudit, + runScopedRepositoryCommand: mocks.runScopedRepositoryCommand, +})) + +import { OperationAdapterExecutionError, type TrustedOperationContext } from '@/worker/operations/executor' +import { productionOperationAdapters } from '@/worker/operations/production-adapters' + +const context: TrustedOperationContext = { + taskId: '11111111-1111-4111-8111-111111111111', + projectId: project.id, + attemptKey: 'attempt-1', + projectRoot: '/trusted/project', + rootBindingRevision: '7', + projectRevision: '2026-08-06T00:00:00.000Z', + policy: { + projectId: project.id, + scopedProjectId: project.id, + policyVersion: 'policy-v1', + allowedCapabilities: ['filesystem.project.read'], + policyCeilings: { repository_read: true }, + }, +} + +describe('production operation adapters', () => { + beforeEach(() => { + vi.clearAllMocks() + mocks.runScopedRepositoryCommand.mockResolvedValue({ + argv: ['status', '--short'], + command: 'git', + cwd: '/trusted/project', + exitCode: 0, + finishedAt: new Date(), + outputSummary: 'M file.ts', + riskClass: 'read_only', + startedAt: new Date(), + stderr: '', + stdout: 'M file.ts', + }) + mocks.recordScopedCommandAudit.mockResolvedValue({ id: '33333333-3333-4333-8333-333333333333' }) + mocks.assertProjectLocalPathForExecution.mockResolvedValue('/trusted/project') + mocks.loadTrustedOperationContext.mockResolvedValue(context) + }) + + it('runs only the fixed Git argv with cancellation and records command evidence', async () => { + const controller = new AbortController() + const result = await productionOperationAdapters.repositoryStatusRead(context, { + signal: controller.signal, + deadline: new Date(Date.now() + 30_000), + }) + expect(mocks.runScopedRepositoryCommand).toHaveBeenCalledWith({ + cwd: '/trusted/project', + command: 'git', + argv: ['status', '--short'], + signal: controller.signal, + }) + expect(mocks.recordScopedCommandAudit).toHaveBeenCalledOnce() + expect(mocks.assertProjectLocalPathForExecution).toHaveBeenCalledWith(project) + expect(result.evidenceRefs).toEqual(['33333333-3333-4333-8333-333333333333']) + }) + + it('uses the fixed current-branch argv for the third cancellable operation', async () => { + await productionOperationAdapters.repositoryBranchRead(context, { + signal: new AbortController().signal, + deadline: new Date(Date.now() + 30_000), + }) + expect(mocks.runScopedRepositoryCommand).toHaveBeenCalledWith(expect.objectContaining({ + command: 'git', + argv: ['branch', '--show-current'], + })) + }) + + it('returns typed command-audit evidence for a nonzero Git exit', async () => { + mocks.runScopedRepositoryCommand.mockResolvedValue({ + argv: ['status', '--short'], + command: 'git', + cwd: '/trusted/project', + exitCode: 1, + finishedAt: new Date(), + outputSummary: 'redacted failure', + riskClass: 'read_only', + startedAt: new Date(), + stderr: 'redacted failure', + stdout: '', + }) + const error = await productionOperationAdapters.repositoryStatusRead(context, { + signal: new AbortController().signal, + deadline: new Date(Date.now() + 30_000), + }).catch((cause: unknown) => cause) + expect(error).toBeInstanceOf(OperationAdapterExecutionError) + expect(error).toMatchObject({ + code: 'adapter_failed', + evidenceRefs: ['33333333-3333-4333-8333-333333333333'], + }) + }) + + it('settles an aborted command audit before returning typed cancellation evidence', async () => { + const controller = new AbortController() + mocks.runScopedRepositoryCommand.mockImplementation(async () => { + controller.abort() + return { + argv: ['status', '--short'], + command: 'git', + cwd: '/trusted/project', + exitCode: 1, + finishedAt: new Date(), + outputSummary: 'cancelled', + riskClass: 'read_only', + startedAt: new Date(), + stderr: '', + stdout: '', + } + }) + const error = await productionOperationAdapters.repositoryStatusRead(context, { + signal: controller.signal, + deadline: new Date(Date.now() + 30_000), + }).catch((cause: unknown) => cause) + expect(error).toBeInstanceOf(OperationAdapterExecutionError) + expect(error).toMatchObject({ + code: 'cancelled', + evidenceRefs: ['33333333-3333-4333-8333-333333333333'], + }) + expect(mocks.recordScopedCommandAudit).toHaveBeenCalledOnce() + }) + + it('fails before the child process when live task/package authority changed', async () => { + mocks.loadTrustedOperationContext.mockResolvedValue({ + ...context, + policy: { ...context.policy, policyVersion: 'different-policy-version' }, + }) + await expect(productionOperationAdapters.repositoryStatusRead(context, { + signal: new AbortController().signal, + deadline: new Date(Date.now() + 30_000), + })).rejects.toThrow('policy changed') + expect(mocks.runScopedRepositoryCommand).not.toHaveBeenCalled() + }) +}) diff --git a/web/__tests__/repository-evidence.test.ts b/web/__tests__/repository-evidence.test.ts index 2a23a643..cdab6a51 100644 --- a/web/__tests__/repository-evidence.test.ts +++ b/web/__tests__/repository-evidence.test.ts @@ -472,13 +472,38 @@ describe('scoped repository command runner', () => { const headDiffResult = await runScopedRepositoryCommand({ cwd: tempRoot, command: 'git', - argv: ['diff', '--stat', 'HEAD', '--'], + argv: ['diff', '--no-ext-diff', '--no-textconv', '--stat', 'HEAD', '--'], }) expect(headDiffResult.riskClass).toBe('read_only') expect(headDiffResult.exitCode).toBe(0) }) + it('disables repository-controlled external diff and textconv execution', async () => { + const sentinel = path.join(tempRoot, 'textconv-ran') + const textconv = path.join(tempRoot, 'malicious-textconv.sh') + await fs.writeFile(textconv, `#!/bin/sh\ntouch "${sentinel}"\ncat "$1"\n`) + await fs.chmod(textconv, 0o700) + await fs.writeFile(path.join(tempRoot, '.gitattributes'), '*.secret diff=malicious\n') + await fs.writeFile(path.join(tempRoot, 'payload.secret'), 'before\n') + await execFile('git', ['config', 'diff.malicious.textconv', textconv], { cwd: tempRoot }) + await execFile('git', ['add', '.gitattributes', 'payload.secret'], { cwd: tempRoot }) + await execFile('git', ['commit', '-m', 'add diff fixture'], { cwd: tempRoot }) + await fs.writeFile(path.join(tempRoot, 'payload.secret'), 'after\n') + + await expect(runScopedRepositoryCommand({ + cwd: tempRoot, + command: 'git', + argv: ['diff', '--no-ext-diff', '--no-textconv', '--stat', '--'], + })).resolves.toMatchObject({ exitCode: 0, riskClass: 'read_only' }) + await expect(fs.stat(sentinel)).rejects.toThrow() + await expect(runScopedRepositoryCommand({ + cwd: tempRoot, + command: 'git', + argv: ['diff', '--stat', '--'], + })).rejects.toThrow(/not allowed/i) + }) + it('detects local validation commands but blocks host package-manager execution', async () => { await fs.writeFile(path.join(tempRoot, 'package.json'), JSON.stringify({ scripts: { diff --git a/web/db/migrations/0030_operation_runs.sql b/web/db/migrations/0030_operation_runs.sql new file mode 100644 index 00000000..0cb486eb --- /dev/null +++ b/web/db/migrations/0030_operation_runs.sql @@ -0,0 +1,190 @@ +-- Bounded deterministic operation ledger. Model-supplied inputs are never +-- persisted; only domain-separated SHA-256 fingerprints cross this boundary. +CREATE TABLE "operation_runs" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "task_id" uuid NOT NULL, + "project_id" uuid NOT NULL, + "work_package_id" uuid, + "agent_run_id" uuid, + "task_attempt_id" uuid, + "execution_outcome_id" uuid, + "definition_schema_version" integer DEFAULT 1 NOT NULL, + "operation_id" text NOT NULL, + "operation_version" integer NOT NULL, + "capability" text NOT NULL, + "idempotency_key" text NOT NULL, + "definition_digest" text NOT NULL, + "scope_fingerprint" text NOT NULL, + "request_fingerprint" text NOT NULL, + "inputs_fingerprint" text NOT NULL, + "reason_fingerprint" text NOT NULL, + "policy_decision" jsonb NOT NULL, + "status" text DEFAULT 'running' NOT NULL, + "verification_status" text DEFAULT 'not_started' NOT NULL, + "output_fingerprint" text, + "outcome_fingerprint" text, + "started_at" timestamp with time zone DEFAULT now() NOT NULL, + "completed_at" timestamp with time zone, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT "operation_runs_task_id_tasks_id_fk" FOREIGN KEY ("task_id") REFERENCES "public"."tasks"("id") ON DELETE restrict ON UPDATE no action, + CONSTRAINT "operation_runs_project_id_projects_id_fk" FOREIGN KEY ("project_id") REFERENCES "public"."projects"("id") ON DELETE restrict ON UPDATE no action, + CONSTRAINT "operation_runs_work_package_id_work_packages_id_fk" FOREIGN KEY ("work_package_id") REFERENCES "public"."work_packages"("id") ON DELETE set null ON UPDATE no action, + CONSTRAINT "operation_runs_agent_run_id_agent_runs_id_fk" FOREIGN KEY ("agent_run_id") REFERENCES "public"."agent_runs"("id") ON DELETE set null ON UPDATE no action, + CONSTRAINT "operation_runs_task_attempt_id_task_attempts_id_fk" FOREIGN KEY ("task_attempt_id") REFERENCES "public"."task_attempts"("id") ON DELETE set null ON UPDATE no action, + CONSTRAINT "operation_runs_execution_outcome_id_execution_outcomes_id_fk" FOREIGN KEY ("execution_outcome_id") REFERENCES "public"."execution_outcomes"("id") ON DELETE restrict ON UPDATE no action, + CONSTRAINT "operation_runs_definition_schema_check" CHECK ("definition_schema_version" = 1), + CONSTRAINT "operation_runs_operation_version_check" CHECK ("operation_version" > 0), + CONSTRAINT "operation_runs_status_check" CHECK ("status" IN ('running', 'completed', 'blocked', 'failed')), + CONSTRAINT "operation_runs_verification_status_check" CHECK ("verification_status" IN ('not_started', 'passed', 'failed')), + CONSTRAINT "operation_runs_fingerprints_check" CHECK ( + "idempotency_key" ~ '^[0-9a-f]{64}$' AND + "definition_digest" ~ '^[0-9a-f]{64}$' AND + "scope_fingerprint" ~ '^[0-9a-f]{64}$' AND + "request_fingerprint" ~ '^[0-9a-f]{64}$' AND + "inputs_fingerprint" ~ '^[0-9a-f]{64}$' AND + "reason_fingerprint" ~ '^[0-9a-f]{64}$' AND + ("output_fingerprint" IS NULL OR "output_fingerprint" ~ '^[0-9a-f]{64}$') AND + ("outcome_fingerprint" IS NULL OR "outcome_fingerprint" ~ '^[0-9a-f]{64}$') + ), + CONSTRAINT "operation_runs_terminal_shape_check" CHECK ( + ("status" = 'running' AND "completed_at" IS NULL AND "execution_outcome_id" IS NULL AND "outcome_fingerprint" IS NULL) OR + ("status" <> 'running' AND "completed_at" IS NOT NULL AND "execution_outcome_id" IS NOT NULL AND "outcome_fingerprint" IS NOT NULL) + ) +); +--> statement-breakpoint +CREATE UNIQUE INDEX "operation_runs_task_idempotency_key_idx" ON "operation_runs" USING btree ("task_id", "idempotency_key"); +--> statement-breakpoint +CREATE INDEX "operation_runs_project_id_created_at_idx" ON "operation_runs" USING btree ("project_id", "created_at"); +--> statement-breakpoint +CREATE INDEX "operation_runs_operation_version_idx" ON "operation_runs" USING btree ("operation_id", "operation_version"); +--> statement-breakpoint +CREATE INDEX "operation_runs_execution_outcome_id_idx" ON "operation_runs" USING btree ("execution_outcome_id"); +--> statement-breakpoint +CREATE TABLE "operation_run_events" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "operation_run_id" uuid NOT NULL, + "sequence" integer NOT NULL, + "phase" text NOT NULL, + "status" text NOT NULL, + "detail_code" text NOT NULL, + "detail_fingerprint" text NOT NULL, + "evidence_refs" jsonb DEFAULT '[]'::jsonb NOT NULL, + "created_at" timestamp with time zone DEFAULT now() NOT NULL, + CONSTRAINT "operation_run_events_operation_run_id_operation_runs_id_fk" FOREIGN KEY ("operation_run_id") REFERENCES "public"."operation_runs"("id") ON DELETE restrict ON UPDATE no action, + CONSTRAINT "operation_run_events_status_check" CHECK ("status" IN ('passed', 'blocked', 'failed')), + CONSTRAINT "operation_run_events_detail_fingerprint_check" CHECK ("detail_fingerprint" ~ '^[0-9a-f]{64}$'), + CONSTRAINT "operation_run_events_evidence_refs_check" CHECK (jsonb_typeof("evidence_refs") = 'array'), + CONSTRAINT "operation_run_events_phase_sequence_check" CHECK ( + ("phase" = 'request_validation' AND "sequence" = 0) OR + ("phase" = 'policy' AND "sequence" = 1) OR + ("phase" = 'preflight' AND "sequence" = 2) OR + ("phase" = 'execution' AND "sequence" = 3) OR + ("phase" = 'verification' AND "sequence" = 4) OR + ("phase" = 'outcome' AND "sequence" = 5) + ) +); +--> statement-breakpoint +CREATE UNIQUE INDEX "operation_run_events_run_sequence_idx" ON "operation_run_events" USING btree ("operation_run_id", "sequence"); +--> statement-breakpoint +CREATE INDEX "operation_run_events_run_created_at_idx" ON "operation_run_events" USING btree ("operation_run_id", "created_at"); +--> statement-breakpoint +CREATE FUNCTION "forge_guard_operation_event_insert_v1"() RETURNS trigger +LANGUAGE plpgsql +SET search_path = pg_catalog, public +AS $$ +DECLARE + v_run_status text; + v_last_sequence integer; + v_last_status text; +BEGIN + SELECT status INTO STRICT v_run_status + FROM public.operation_runs + WHERE id = NEW.operation_run_id + FOR UPDATE; + IF v_run_status <> 'running' THEN + RAISE EXCEPTION 'terminal operation runs cannot receive events'; + END IF; + SELECT sequence, status INTO v_last_sequence, v_last_status + FROM public.operation_run_events + WHERE operation_run_id = NEW.operation_run_id + ORDER BY sequence DESC + LIMIT 1; + IF (v_last_sequence IS NULL AND NEW.sequence <> 0) + OR (v_last_sequence = 0 AND (v_last_status <> 'passed' OR NEW.sequence <> 1)) + OR (v_last_sequence = 1 AND ( + (v_last_status = 'passed' AND NEW.sequence <> 2) + OR (v_last_status <> 'passed' AND NEW.sequence <> 5) + )) + OR (v_last_sequence = 2 AND ( + (v_last_status = 'passed' AND NEW.sequence <> 3) + OR (v_last_status <> 'passed' AND NEW.sequence <> 5) + )) + OR (v_last_sequence = 3 AND ( + (v_last_status = 'passed' AND NEW.sequence <> 4) + OR (v_last_status <> 'passed' AND NEW.sequence <> 5) + )) + OR (v_last_sequence = 4 AND NEW.sequence <> 5) + OR v_last_sequence = 5 THEN + RAISE EXCEPTION 'operation run events must be appended in phase order'; + END IF; + RETURN NEW; +END; +$$; +--> statement-breakpoint +CREATE TRIGGER "operation_run_events_order_guard" +BEFORE INSERT ON "operation_run_events" +FOR EACH ROW EXECUTE FUNCTION "forge_guard_operation_event_insert_v1"(); +--> statement-breakpoint +CREATE FUNCTION "forge_guard_operation_run_history_v1"() RETURNS trigger +LANGUAGE plpgsql +SET search_path = pg_catalog, public +AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + RAISE EXCEPTION 'operation run history is append-only'; + END IF; + IF OLD.status <> 'running' THEN + RAISE EXCEPTION 'terminal operation runs are immutable'; + END IF; + IF NEW.status = 'running' + OR NEW.id IS DISTINCT FROM OLD.id + OR NEW.task_id IS DISTINCT FROM OLD.task_id + OR NEW.project_id IS DISTINCT FROM OLD.project_id + OR NEW.work_package_id IS DISTINCT FROM OLD.work_package_id + OR NEW.agent_run_id IS DISTINCT FROM OLD.agent_run_id + OR NEW.task_attempt_id IS DISTINCT FROM OLD.task_attempt_id + OR NEW.definition_schema_version IS DISTINCT FROM OLD.definition_schema_version + OR NEW.operation_id IS DISTINCT FROM OLD.operation_id + OR NEW.operation_version IS DISTINCT FROM OLD.operation_version + OR NEW.capability IS DISTINCT FROM OLD.capability + OR NEW.idempotency_key IS DISTINCT FROM OLD.idempotency_key + OR NEW.definition_digest IS DISTINCT FROM OLD.definition_digest + OR NEW.scope_fingerprint IS DISTINCT FROM OLD.scope_fingerprint + OR NEW.request_fingerprint IS DISTINCT FROM OLD.request_fingerprint + OR NEW.inputs_fingerprint IS DISTINCT FROM OLD.inputs_fingerprint + OR NEW.reason_fingerprint IS DISTINCT FROM OLD.reason_fingerprint + OR NEW.policy_decision IS DISTINCT FROM OLD.policy_decision + OR NEW.started_at IS DISTINCT FROM OLD.started_at + OR NEW.created_at IS DISTINCT FROM OLD.created_at THEN + RAISE EXCEPTION 'operation run identity is immutable'; + END IF; + RETURN NEW; +END; +$$; +--> statement-breakpoint +CREATE TRIGGER "operation_runs_history_guard" +BEFORE UPDATE OR DELETE ON "operation_runs" +FOR EACH ROW EXECUTE FUNCTION "forge_guard_operation_run_history_v1"(); +--> statement-breakpoint +CREATE FUNCTION "forge_reject_operation_event_mutation_v1"() RETURNS trigger +LANGUAGE plpgsql +SET search_path = pg_catalog, public +AS $$ +BEGIN + RAISE EXCEPTION 'operation run events are append-only'; +END; +$$; +--> statement-breakpoint +CREATE TRIGGER "operation_run_events_append_only" +BEFORE UPDATE OR DELETE ON "operation_run_events" +FOR EACH ROW EXECUTE FUNCTION "forge_reject_operation_event_mutation_v1"(); diff --git a/web/db/migrations/meta/_journal.json b/web/db/migrations/meta/_journal.json index d10adedd..0c89d3b6 100644 --- a/web/db/migrations/meta/_journal.json +++ b/web/db/migrations/meta/_journal.json @@ -211,6 +211,13 @@ "when": 1785820800000, "tag": "0029_execution_outcomes", "breakpoints": true + }, + { + "idx": 30, + "version": "7", + "when": 1785993600000, + "tag": "0030_operation_runs", + "breakpoints": true } ] } diff --git a/web/db/schema.ts b/web/db/schema.ts index cf65def9..1da5caa3 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -1499,6 +1499,104 @@ export const executionOutcomes = pgTable( export type ExecutionOutcomeRow = InferSelectModel export type NewExecutionOutcomeRow = InferInsertModel +// --------------------------------------------------------------------------- +// operationRuns and operationRunEvents +// --------------------------------------------------------------------------- +// A run has one immutable identity/idempotency tuple and is terminalized once. +// Detailed phase history is kept in the append-only event table below. +export const operationRuns = pgTable( + 'operation_runs', + { + id: uuid('id').primaryKey().defaultRandom(), + taskId: uuid('task_id').notNull().references(() => tasks.id, { onDelete: 'restrict' }), + projectId: uuid('project_id').notNull().references(() => projects.id, { onDelete: 'restrict' }), + workPackageId: uuid('work_package_id').references(() => workPackages.id, { onDelete: 'set null' }), + agentRunId: uuid('agent_run_id').references(() => agentRuns.id, { onDelete: 'set null' }), + taskAttemptId: uuid('task_attempt_id').references(() => taskAttempts.id, { onDelete: 'set null' }), + executionOutcomeId: uuid('execution_outcome_id').references(() => executionOutcomes.id, { onDelete: 'restrict' }), + definitionSchemaVersion: integer('definition_schema_version').notNull().default(1), + operationId: text('operation_id').notNull(), + operationVersion: integer('operation_version').notNull(), + capability: text('capability').notNull(), + idempotencyKey: text('idempotency_key').notNull(), + definitionDigest: text('definition_digest').notNull(), + scopeFingerprint: text('scope_fingerprint').notNull(), + requestFingerprint: text('request_fingerprint').notNull(), + inputsFingerprint: text('inputs_fingerprint').notNull(), + reasonFingerprint: text('reason_fingerprint').notNull(), + policyDecision: jsonb('policy_decision').$type>().notNull(), + status: text('status').notNull().default('running'), + verificationStatus: text('verification_status').notNull().default('not_started'), + outputFingerprint: text('output_fingerprint'), + outcomeFingerprint: text('outcome_fingerprint'), + startedAt: timestamp('started_at', tsOpts).defaultNow().notNull(), + completedAt: timestamp('completed_at', tsOpts), + createdAt: timestamp('created_at', tsOpts).defaultNow().notNull(), + }, + (t) => [ + uniqueIndex('operation_runs_task_idempotency_key_idx').on(t.taskId, t.idempotencyKey), + index('operation_runs_project_id_created_at_idx').on(t.projectId, t.createdAt), + index('operation_runs_operation_version_idx').on(t.operationId, t.operationVersion), + index('operation_runs_execution_outcome_id_idx').on(t.executionOutcomeId), + check('operation_runs_definition_schema_check', sql`${t.definitionSchemaVersion} = 1`), + check('operation_runs_operation_version_check', sql`${t.operationVersion} > 0`), + check('operation_runs_status_check', sql`${t.status} IN ('running', 'completed', 'blocked', 'failed')`), + check('operation_runs_verification_status_check', sql`${t.verificationStatus} IN ('not_started', 'passed', 'failed')`), + check('operation_runs_fingerprints_check', sql` + ${t.idempotencyKey} ~ '^[0-9a-f]{64}$' AND + ${t.definitionDigest} ~ '^[0-9a-f]{64}$' AND + ${t.scopeFingerprint} ~ '^[0-9a-f]{64}$' AND + ${t.requestFingerprint} ~ '^[0-9a-f]{64}$' AND + ${t.inputsFingerprint} ~ '^[0-9a-f]{64}$' AND + ${t.reasonFingerprint} ~ '^[0-9a-f]{64}$' AND + (${t.outputFingerprint} IS NULL OR ${t.outputFingerprint} ~ '^[0-9a-f]{64}$') + AND (${t.outcomeFingerprint} IS NULL OR ${t.outcomeFingerprint} ~ '^[0-9a-f]{64}$') + `), + check('operation_runs_terminal_shape_check', sql` + (${t.status} = 'running' AND ${t.completedAt} IS NULL AND ${t.executionOutcomeId} IS NULL AND ${t.outcomeFingerprint} IS NULL) OR + (${t.status} <> 'running' AND ${t.completedAt} IS NOT NULL AND ${t.executionOutcomeId} IS NOT NULL AND ${t.outcomeFingerprint} IS NOT NULL) + `), + ], +) + +export type OperationRun = InferSelectModel +export type NewOperationRun = InferInsertModel + +export const operationRunEvents = pgTable( + 'operation_run_events', + { + id: uuid('id').primaryKey().defaultRandom(), + operationRunId: uuid('operation_run_id') + .notNull() + .references(() => operationRuns.id, { onDelete: 'restrict' }), + sequence: integer('sequence').notNull(), + phase: text('phase').notNull(), + status: text('status').notNull(), + detailCode: text('detail_code').notNull(), + detailFingerprint: text('detail_fingerprint').notNull(), + evidenceRefs: jsonb('evidence_refs').$type().notNull().default(sql`'[]'::jsonb`), + createdAt: timestamp('created_at', tsOpts).defaultNow().notNull(), + }, + (t) => [ + uniqueIndex('operation_run_events_run_sequence_idx').on(t.operationRunId, t.sequence), + index('operation_run_events_run_created_at_idx').on(t.operationRunId, t.createdAt), + check('operation_run_events_status_check', sql`${t.status} IN ('passed', 'blocked', 'failed')`), + check('operation_run_events_detail_fingerprint_check', sql`${t.detailFingerprint} ~ '^[0-9a-f]{64}$'`), + check('operation_run_events_evidence_refs_check', sql`jsonb_typeof(${t.evidenceRefs}) = 'array'`), + check('operation_run_events_phase_sequence_check', sql` + (${t.phase} = 'request_validation' AND ${t.sequence} = 0) OR + (${t.phase} = 'policy' AND ${t.sequence} = 1) OR + (${t.phase} = 'preflight' AND ${t.sequence} = 2) OR + (${t.phase} = 'execution' AND ${t.sequence} = 3) OR + (${t.phase} = 'verification' AND ${t.sequence} = 4) OR + (${t.phase} = 'outcome' AND ${t.sequence} = 5) + `), + ], +) + +export type OperationRunEvent = InferSelectModel +export type NewOperationRunEvent = InferInsertModel + // --------------------------------------------------------------------------- // artifacts // --------------------------------------------------------------------------- diff --git a/web/lib/operations/catalog.ts b/web/lib/operations/catalog.ts new file mode 100644 index 00000000..f1c2fd33 --- /dev/null +++ b/web/lib/operations/catalog.ts @@ -0,0 +1,124 @@ +import { + OPERATION_DEFINITION_SCHEMA_VERSION, + OperationContractError, + hasExactKeys, + type OperationDefinition, + type OperationRequest, +} from './contracts' + +const EMPTY_INPUT_KEYS = [] as const + +export const BUILT_IN_OPERATIONS = [ + { + schemaVersion: OPERATION_DEFINITION_SCHEMA_VERSION, + id: 'repository.status.read', + version: 1, + description: 'Read the current bounded Git repository status for the trusted project.', + capability: 'filesystem.project.read', + risk: 'read_only', + inputKeys: EMPTY_INPUT_KEYS, + scope: 'trusted_project', + requiredPolicyCeiling: 'repository_read', + adapter: 'repository_status_read_v1', + timeoutMs: 30_000, + verification: 'deterministic_adapter', + recovery: 'none_read_only', + approvalRequired: false, + independentVerificationRequired: false, + audit: { persistInputs: false, fingerprint: 'sha256', reasonUse: 'audit_only' }, + enabled: true, + deprecated: false, + }, + { + schemaVersion: OPERATION_DEFINITION_SCHEMA_VERSION, + id: 'repository.diff.summary', + version: 1, + description: 'Read a bounded diff-stat summary for the trusted project working tree.', + capability: 'filesystem.project.read', + risk: 'read_only', + inputKeys: EMPTY_INPUT_KEYS, + scope: 'trusted_project', + requiredPolicyCeiling: 'repository_read', + adapter: 'repository_diff_summary_v1', + timeoutMs: 30_000, + verification: 'deterministic_adapter', + recovery: 'none_read_only', + approvalRequired: false, + independentVerificationRequired: false, + audit: { persistInputs: false, fingerprint: 'sha256', reasonUse: 'audit_only' }, + enabled: true, + deprecated: false, + }, + { + schemaVersion: OPERATION_DEFINITION_SCHEMA_VERSION, + id: 'repository.branch.read', + version: 1, + description: 'Read the current branch name for the trusted Git repository.', + capability: 'filesystem.project.read', + risk: 'read_only', + inputKeys: EMPTY_INPUT_KEYS, + scope: 'trusted_project', + requiredPolicyCeiling: 'repository_read', + adapter: 'repository_branch_read_v1', + timeoutMs: 30_000, + verification: 'deterministic_adapter', + recovery: 'none_read_only', + approvalRequired: false, + independentVerificationRequired: false, + audit: { persistInputs: false, fingerprint: 'sha256', reasonUse: 'audit_only' }, + enabled: true, + deprecated: false, + }, +] as const satisfies readonly OperationDefinition[] + +function definitionKey(definition: Pick): string { + return `${definition.id}@${definition.version}` +} + +export function createOperationCatalog( + definitions: readonly OperationDefinition[], +): ReadonlyMap { + const catalog = new Map() + for (const definition of definitions) { + if (definition.schemaVersion !== OPERATION_DEFINITION_SCHEMA_VERSION) { + throw new Error(`Unsupported operation definition schema for ${definition.id}.`) + } + if (!Number.isSafeInteger(definition.version) || definition.version < 1) { + throw new Error(`Invalid operation version for ${definition.id}.`) + } + if (definition.risk !== 'read_only' || definition.recovery !== 'none_read_only') { + throw new Error(`The v1 operation catalog only accepts read-only operations: ${definition.id}.`) + } + if (definition.audit.persistInputs || definition.audit.reasonUse !== 'audit_only') { + throw new Error(`The v1 operation audit contract is not safe for ${definition.id}.`) + } + const key = definitionKey(definition) + if (catalog.has(key)) throw new Error(`Duplicate operation definition: ${key}.`) + catalog.set(key, Object.freeze({ ...definition, inputKeys: Object.freeze([...definition.inputKeys]) })) + } + return catalog +} + +export const OPERATION_CATALOG = createOperationCatalog(BUILT_IN_OPERATIONS) + +export function resolveOperationDefinition( + request: Pick, + catalog: ReadonlyMap = OPERATION_CATALOG, +): OperationDefinition { + const definition = catalog.get(`${request.operationId}@${request.operationVersion}`) + if (definition) return definition + const knownId = [...catalog.values()].some((candidate) => candidate.id === request.operationId) + throw new OperationContractError( + knownId ? 'unsupported_version' : 'unknown_operation', + knownId ? 'Requested operation version is not supported.' : 'Requested operation is not registered.', + ) +} + +export function validateOperationInputs( + definition: OperationDefinition, + inputs: Record, +): void { + if (!hasExactKeys(inputs, definition.inputKeys)) { + throw new OperationContractError('invalid_request', `Inputs do not match ${definition.id}@${definition.version}.`) + } +} diff --git a/web/lib/operations/contracts.ts b/web/lib/operations/contracts.ts new file mode 100644 index 00000000..49c8e600 --- /dev/null +++ b/web/lib/operations/contracts.ts @@ -0,0 +1,177 @@ +import { createHash } from 'node:crypto' + +import type { ExecutionOutcome } from '@/lib/execution-outcomes' + +export const OPERATION_DEFINITION_SCHEMA_VERSION = 1 as const +export const OPERATION_REQUEST_KEYS = [ + 'inputs', + 'operationId', + 'operationVersion', + 'reason', + 'requestedCapability', + 'schemaVersion', +] as const + +export const OPERATION_PHASES = [ + 'request_validation', + 'policy', + 'preflight', + 'execution', + 'verification', + 'outcome', +] as const + +export type OperationPhase = typeof OPERATION_PHASES[number] +export type OperationRunStatus = 'running' | 'completed' | 'blocked' | 'failed' +export type OperationVerificationStatus = 'not_started' | 'passed' | 'failed' +export type OperationAdapterKind = + | 'repository_status_read_v1' + | 'repository_diff_summary_v1' + | 'repository_branch_read_v1' +export type OperationPolicyCeiling = 'repository_read' + +export type OperationRequest = { + schemaVersion: 1 + operationId: string + operationVersion: number + inputs: Record + reason: string + requestedCapability: string +} + +export type OperationDefinition = { + schemaVersion: 1 + id: string + version: number + description: string + capability: string + risk: 'read_only' + inputKeys: readonly string[] + scope: 'trusted_project' + requiredPolicyCeiling: OperationPolicyCeiling + adapter: OperationAdapterKind + timeoutMs: number + verification: 'deterministic_adapter' + recovery: 'none_read_only' + approvalRequired: false + independentVerificationRequired: false + audit: { + persistInputs: false + fingerprint: 'sha256' + reasonUse: 'audit_only' + } + enabled: boolean + deprecated: boolean +} + +export type OperationPolicyDecision = { + schemaVersion: 1 + allowed: boolean + code: + | 'allowed' + | 'capability_mismatch' + | 'missing_capability' + | 'scope_mismatch' + | 'policy_ceiling_denied' + | 'operation_disabled' + | 'operation_deprecated' + capability: string + projectId: string + policyVersion: string +} + +export type OperationExecutionResult = { + runId: string + operationId: string + operationVersion: number + definitionDigest: string + scopeFingerprint: string + status: OperationRunStatus + verificationStatus: OperationVerificationStatus + replayed: boolean + output: unknown | null + outcome: ExecutionOutcome | null +} + +export function isSha256Fingerprint(value: unknown): value is string { + return typeof value === 'string' && /^[0-9a-f]{64}$/.test(value) +} + +export class OperationContractError extends Error { + readonly code: 'invalid_request' | 'unknown_operation' | 'unsupported_version' + + constructor( + code: OperationContractError['code'], + message: string, + ) { + super(message) + this.name = 'OperationContractError' + this.code = code + } +} + +export function isPlainRecord(value: unknown): value is Record { + if (typeof value !== 'object' || value === null || Array.isArray(value)) return false + const prototype = Object.getPrototypeOf(value) + return prototype === Object.prototype || prototype === null +} + +export function hasExactKeys(value: Record, expected: readonly string[]): boolean { + const actual = Object.keys(value).sort() + const required = [...expected].sort() + return actual.length === required.length && actual.every((key, index) => key === required[index]) +} + +export function parseOperationRequest(value: unknown): OperationRequest { + if (!isPlainRecord(value) || !hasExactKeys(value, OPERATION_REQUEST_KEYS)) { + throw new OperationContractError('invalid_request', 'Operation request must contain exactly the v1 request keys.') + } + if (value.schemaVersion !== 1) { + throw new OperationContractError('invalid_request', 'Operation request schema version must be 1.') + } + if (typeof value.operationId !== 'string' || value.operationId.length > 200 || !/^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)+$/.test(value.operationId)) { + throw new OperationContractError('invalid_request', 'Operation id is invalid.') + } + if (!Number.isSafeInteger(value.operationVersion) || (value.operationVersion as number) < 1) { + throw new OperationContractError('invalid_request', 'Operation version must be a positive integer.') + } + if (!isPlainRecord(value.inputs)) { + throw new OperationContractError('invalid_request', 'Operation inputs must be a plain object.') + } + if (typeof value.reason !== 'string' || value.reason.trim().length < 1 || value.reason.length > 500 || /[\u0000-\u0008\u000b\u000c\u000e-\u001f\u007f]/u.test(value.reason)) { + throw new OperationContractError('invalid_request', 'Operation reason must be printable text between 1 and 500 characters.') + } + if (typeof value.requestedCapability !== 'string' || value.requestedCapability.length > 200 || !/^[a-z][a-z0-9]*(?:[._-][a-z0-9]+)+$/.test(value.requestedCapability)) { + throw new OperationContractError('invalid_request', 'Requested capability is invalid.') + } + + return { + schemaVersion: 1, + operationId: value.operationId, + operationVersion: value.operationVersion as number, + inputs: value.inputs, + reason: value.reason, + requestedCapability: value.requestedCapability, + } +} + +/** Canonical JSON for hashes only. Values are never reconstructed from it. */ +export function canonicalJson(value: unknown): string { + if (value === null || typeof value === 'boolean' || typeof value === 'string') return JSON.stringify(value) + if (typeof value === 'number') { + if (!Number.isFinite(value)) throw new Error('Non-finite values cannot be fingerprinted.') + return JSON.stringify(value) + } + if (Array.isArray(value)) return `[${value.map(canonicalJson).join(',')}]` + if (isPlainRecord(value)) { + return `{${Object.keys(value).sort().map((key) => `${JSON.stringify(key)}:${canonicalJson(value[key])}`).join(',')}}` + } + throw new Error('Only JSON values can be fingerprinted.') +} + +export function operationFingerprint(domain: string, value: unknown): string { + return createHash('sha256') + .update(`forge:operation:${domain}:v1\0`, 'utf8') + .update(canonicalJson(value), 'utf8') + .digest('hex') +} diff --git a/web/lib/operations/policy.ts b/web/lib/operations/policy.ts new file mode 100644 index 00000000..c10bbe0f --- /dev/null +++ b/web/lib/operations/policy.ts @@ -0,0 +1,37 @@ +import type { OperationDefinition, OperationPolicyCeiling, OperationPolicyDecision } from './contracts' + +export type TrustedOperationPolicyContext = { + projectId: string + scopedProjectId: string + policyVersion: string + allowedCapabilities: readonly string[] + policyCeilings: Readonly> +} + +export function evaluateOperationPolicy(input: { + definition: OperationDefinition + requestedCapability: string + context: TrustedOperationPolicyContext +}): OperationPolicyDecision { + const base = { + schemaVersion: 1 as const, + capability: input.definition.capability, + projectId: input.context.projectId, + policyVersion: input.context.policyVersion, + } + if (!input.definition.enabled) return { ...base, allowed: false, code: 'operation_disabled' } + if (input.definition.deprecated) return { ...base, allowed: false, code: 'operation_deprecated' } + if (input.requestedCapability !== input.definition.capability) { + return { ...base, allowed: false, code: 'capability_mismatch' } + } + if (input.context.projectId !== input.context.scopedProjectId) { + return { ...base, allowed: false, code: 'scope_mismatch' } + } + if (!input.context.allowedCapabilities.includes(input.definition.capability)) { + return { ...base, allowed: false, code: 'missing_capability' } + } + if (input.context.policyCeilings[input.definition.requiredPolicyCeiling] !== true) { + return { ...base, allowed: false, code: 'policy_ceiling_denied' } + } + return { ...base, allowed: true, code: 'allowed' } +} diff --git a/web/scripts/ci/prove-installer-legacy-migration-repair.sh b/web/scripts/ci/prove-installer-legacy-migration-repair.sh index 035b995a..24c5b312 100644 --- a/web/scripts/ci/prove-installer-legacy-migration-repair.sh +++ b/web/scripts/ci/prove-installer-legacy-migration-repair.sh @@ -1307,8 +1307,8 @@ assert_unchanged managed-latest-once managed-latest-twice 'Managed latest rerun' admin_psql <<'SQL' DO $proof$ BEGIN - IF (SELECT count(*) FROM drizzle.__drizzle_migrations) <> 30 - OR (SELECT max(created_at) FROM drizzle.__drizzle_migrations) <> 1785820800000 THEN + IF (SELECT count(*) FROM drizzle.__drizzle_migrations) <> 31 + OR (SELECT max(created_at) FROM drizzle.__drizzle_migrations) <> 1785993600000 THEN RAISE EXCEPTION 'Managed sequence did not reach the exact latest ledger'; END IF; IF EXISTS ( diff --git a/web/scripts/ci/sql/migration-0027-expansion-assertions.sql b/web/scripts/ci/sql/migration-0027-expansion-assertions.sql index 70971c77..1aeff55e 100644 --- a/web/scripts/ci/sql/migration-0027-expansion-assertions.sql +++ b/web/scripts/ci/sql/migration-0027-expansion-assertions.sql @@ -13,11 +13,11 @@ BEGIN -- role.rolpassword IS NULL is verified by the administrator-only S4 -- bootstrap; pg_roles intentionally masks it from this ordinary migration -- proof. This block verifies every attribute visible to the ordinary login. - IF (SELECT count(*) FROM drizzle.__drizzle_migrations) <> 30 - OR (SELECT count(DISTINCT created_at) FROM drizzle.__drizzle_migrations) <> 30 + IF (SELECT count(*) FROM drizzle.__drizzle_migrations) <> 31 + OR (SELECT count(DISTINCT created_at) FROM drizzle.__drizzle_migrations) <> 31 OR NOT EXISTS (SELECT 1 FROM drizzle.__drizzle_migrations WHERE created_at = 1784270400000) OR NOT EXISTS (SELECT 1 FROM drizzle.__drizzle_migrations WHERE created_at = 1784274000000) - OR (SELECT max(created_at) FROM drizzle.__drizzle_migrations) <> 1785820800000 THEN + OR (SELECT max(created_at) FROM drizzle.__drizzle_migrations) <> 1785993600000 THEN RAISE EXCEPTION 'The normal migrator did not retain the immutable 0027/0028 release boundary while reaching the exact latest ledger'; END IF; diff --git a/web/worker/operations/adapters/repository-read.ts b/web/worker/operations/adapters/repository-read.ts new file mode 100644 index 00000000..3487891b --- /dev/null +++ b/web/worker/operations/adapters/repository-read.ts @@ -0,0 +1,78 @@ +import path from 'node:path' + +import type { ScopedCommandResult } from '@/worker/repository-evidence' +import type { OperationAdapterExecution } from '../executor' + +export type RepositoryReadCommand = (input: { + cwd: string + command: 'git' + argv: string[] + signal?: AbortSignal +}) => Promise> + +export type RepositoryReadAdapterResult = { + output: { + exitCode: number + summary: string + } + evidenceRefs: string[] +} + +function assertTrustedProjectRoot(projectRoot: string): string { + if (!path.isAbsolute(projectRoot) || projectRoot.includes('\0')) { + throw new Error('The trusted project root must be an absolute filesystem path.') + } + return path.resolve(/* turbopackIgnore: true */ projectRoot) +} + +async function runFixedGitRead(input: { + projectRoot: string + argv: readonly string[] + runCommand: RepositoryReadCommand + execution?: OperationAdapterExecution +}): Promise { + const result = await input.runCommand({ + cwd: assertTrustedProjectRoot(input.projectRoot), + command: 'git', + argv: [...input.argv], + ...(input.execution ? { signal: input.execution.signal } : {}), + }) + if (result.exitCode !== 0) throw new Error('The fixed repository read did not complete successfully.') + return { + output: { exitCode: result.exitCode, summary: result.outputSummary }, + evidenceRefs: [], + } +} + +export function createRepositoryStatusReadAdapter(input: { + runCommand: RepositoryReadCommand +}): (context: { projectRoot: string | null }, execution?: OperationAdapterExecution) => Promise { + return (context, execution) => runFixedGitRead({ + projectRoot: context.projectRoot ?? '', + argv: ['status', '--short'], + runCommand: input.runCommand, + execution, + }) +} + +export function createRepositoryDiffSummaryAdapter(input: { + runCommand: RepositoryReadCommand +}): (context: { projectRoot: string | null }, execution?: OperationAdapterExecution) => Promise { + return (context, execution) => runFixedGitRead({ + projectRoot: context.projectRoot ?? '', + argv: ['diff', '--no-ext-diff', '--no-textconv', '--stat', '--'], + runCommand: input.runCommand, + execution, + }) +} + +export function createRepositoryBranchReadAdapter(input: { + runCommand: RepositoryReadCommand +}): (context: { projectRoot: string | null }, execution?: OperationAdapterExecution) => Promise { + return (context, execution) => runFixedGitRead({ + projectRoot: context.projectRoot ?? '', + argv: ['branch', '--show-current'], + runCommand: input.runCommand, + execution, + }) +} diff --git a/web/worker/operations/context.ts b/web/worker/operations/context.ts new file mode 100644 index 00000000..bab76404 --- /dev/null +++ b/web/worker/operations/context.ts @@ -0,0 +1,272 @@ +import { eq } from 'drizzle-orm' + +import { db } from '@/db' +import { agentRuns, projects, taskAttempts, tasks, workPackages } from '@/db/schema' +import { operationFingerprint } from '@/lib/operations/contracts' +import { readEffectiveGrantState } from '@/lib/mcps/admission' +import { loadCurrentProjectFilesystemDecision } from '@/lib/mcps/filesystem-grant-reconciliation' +import { approvedEffectiveFilesystemCapabilities } from '@/lib/mcps/filesystem-grants' +import { assertProjectLocalPathForExecution } from '@/lib/projects/local-path' +import { + executeOperation, + type TrustedOperationContext, +} from './executor' +import { databaseOperationLedger } from './ledger' + +type JoinedTaskProject = { + taskId: string + taskStatus: string + taskUpdatedAt: Date + projectId: string + localPath: string | null + grantDecisionRevision: bigint + rootBindingRevision: bigint + projectUpdatedAt: Date + projectMcpConfig: unknown +} + +export type TrustedOperationContextLoaderDependencies = { + loadTaskProject: (taskId: string) => Promise + canonicalizeProjectRoot: (project: { id: string; localPath: string | null }) => Promise + loadProjectFilesystemDecision?: (projectId: string) => Promise + loadLinkedIds?: (input: { + taskId: string + workPackageId: string | null + agentRunId: string | null + taskAttemptId: string | null + }) => Promise<{ + workPackageId: string | null + workPackageMetadata: unknown | null + workPackageUpdatedAt: Date | null + agentRunId: string | null + taskAttemptId: string | null + } | null> +} + +async function loadTaskProjectFromDatabase(taskId: string): Promise { + const [row] = await db + .select({ + taskId: tasks.id, + taskStatus: tasks.status, + taskUpdatedAt: tasks.updatedAt, + projectId: projects.id, + localPath: projects.localPath, + grantDecisionRevision: projects.grantDecisionRevision, + rootBindingRevision: projects.rootBindingRevision, + projectUpdatedAt: projects.updatedAt, + projectMcpConfig: projects.mcpConfig, + }) + .from(tasks) + .innerJoin(projects, eq(tasks.projectId, projects.id)) + .where(eq(tasks.id, taskId)) + .limit(1) + return row ?? null +} + +async function loadLinkedIdsFromDatabase(input: { + taskId: string + workPackageId: string | null + agentRunId: string | null + taskAttemptId: string | null +}) { + const [workPackageRows, agentRunRows, taskAttemptRows] = await Promise.all([ + input.workPackageId + ? db.select({ + id: workPackages.id, + taskId: workPackages.taskId, + metadata: workPackages.metadata, + updatedAt: workPackages.updatedAt, + }) + .from(workPackages).where(eq(workPackages.id, input.workPackageId)).limit(1) + : Promise.resolve([]), + input.agentRunId + ? db.select({ id: agentRuns.id, taskId: agentRuns.taskId, workPackageId: agentRuns.workPackageId }) + .from(agentRuns).where(eq(agentRuns.id, input.agentRunId)).limit(1) + : Promise.resolve([]), + input.taskAttemptId + ? db.select({ id: taskAttempts.id, taskId: taskAttempts.taskId }) + .from(taskAttempts).where(eq(taskAttempts.id, input.taskAttemptId)).limit(1) + : Promise.resolve([]), + ]) + const workPackage = workPackageRows[0] ?? null + const agentRun = agentRunRows[0] ?? null + const taskAttempt = taskAttemptRows[0] ?? null + if ( + (input.workPackageId && workPackage?.taskId !== input.taskId) + || (input.agentRunId && agentRun?.taskId !== input.taskId) + || (input.taskAttemptId && taskAttempt?.taskId !== input.taskId) + || (input.workPackageId && input.agentRunId && agentRun?.workPackageId !== input.workPackageId) + ) { + return null + } + if (agentRun?.workPackageId && !input.workPackageId) { + const [agentWorkPackage] = await db + .select({ + taskId: workPackages.taskId, + metadata: workPackages.metadata, + updatedAt: workPackages.updatedAt, + }) + .from(workPackages) + .where(eq(workPackages.id, agentRun.workPackageId)) + .limit(1) + if (agentWorkPackage?.taskId !== input.taskId) return null + return { + workPackageId: agentRun.workPackageId, + workPackageMetadata: agentWorkPackage.metadata, + workPackageUpdatedAt: agentWorkPackage.updatedAt, + agentRunId: input.agentRunId, + taskAttemptId: input.taskAttemptId, + } + } + return { + workPackageId: input.workPackageId ?? agentRun?.workPackageId ?? null, + workPackageMetadata: workPackage?.metadata ?? null, + workPackageUpdatedAt: workPackage?.updatedAt ?? null, + agentRunId: input.agentRunId, + taskAttemptId: input.taskAttemptId, + } +} + +const productionContextDependencies: TrustedOperationContextLoaderDependencies = { + loadTaskProject: loadTaskProjectFromDatabase, + canonicalizeProjectRoot: assertProjectLocalPathForExecution, + loadProjectFilesystemDecision: loadCurrentProjectFilesystemDecision, + loadLinkedIds: loadLinkedIdsFromDatabase, +} + +/** + * Loads task and project scope from PostgreSQL. Callers cannot supply or + * override the project id, repository root, or project revisions. + */ +export type TrustedOperationContextInput = { + taskId: string + attemptKey: string + workPackageId?: string | null + agentRunId?: string | null + taskAttemptId?: string | null +} + +async function loadTrustedOperationContextWithDependencies( + input: TrustedOperationContextInput, + dependencies: TrustedOperationContextLoaderDependencies, +): Promise { + const joined = await dependencies.loadTaskProject(input.taskId) + if (!joined || joined.taskId !== input.taskId) { + throw new Error('Operation task and project context could not be resolved.') + } + if (joined.taskStatus !== 'approved' && joined.taskStatus !== 'running') { + throw new Error('Deterministic operations require a current approved or running task.') + } + const projectRoot = joined.localPath?.trim() + ? await dependencies.canonicalizeProjectRoot({ id: joined.projectId, localPath: joined.localPath }) + : null + const requestedLinks = { + taskId: input.taskId, + workPackageId: input.workPackageId ?? null, + agentRunId: input.agentRunId ?? null, + taskAttemptId: input.taskAttemptId ?? null, + } + const hasLinkedId = Boolean(requestedLinks.workPackageId || requestedLinks.agentRunId || requestedLinks.taskAttemptId) + const linked = hasLinkedId + ? await dependencies.loadLinkedIds?.(requestedLinks) ?? null + : { + ...requestedLinks, + workPackageMetadata: null, + workPackageUpdatedAt: null, + } + if (!linked) throw new Error('Operation run links do not belong to the authoritative task context.') + const projectFilesystemDecision = await dependencies.loadProjectFilesystemDecision?.(joined.projectId) ?? null + const filesystemCapabilities = approvedEffectiveFilesystemCapabilities(linked.workPackageMetadata) + const effectiveFilesystemGrant = readEffectiveGrantState( + { metadata: linked.workPackageMetadata }, + { + mcpConfig: joined.projectMcpConfig, + filesystemGrantDecision: projectFilesystemDecision, + rootBindingRevision: joined.rootBindingRevision, + }, + ['filesystem.project.read'], + ) + const repositoryReadAllowed = Boolean( + linked.workPackageId + && filesystemCapabilities.includes('filesystem.project.read') + && effectiveFilesystemGrant.status === 'approved' + && effectiveFilesystemGrant.grantMode === 'always_allow' + && effectiveFilesystemGrant.coveredCapabilities.includes('filesystem.project.read'), + ) + const policyVersion = operationFingerprint('authoritative-policy', { + taskStatus: joined.taskStatus, + taskRevision: joined.taskUpdatedAt.toISOString(), + projectRevision: joined.projectUpdatedAt.toISOString(), + grantDecisionRevision: joined.grantDecisionRevision.toString(), + rootBindingRevision: joined.rootBindingRevision.toString(), + projectMcpConfig: joined.projectMcpConfig, + projectFilesystemDecision, + workPackageId: linked.workPackageId, + workPackageRevision: linked.workPackageUpdatedAt?.toISOString() ?? null, + filesystemCapabilities, + }) + + return { + taskId: joined.taskId, + projectId: joined.projectId, + attemptKey: input.attemptKey, + projectRoot, + rootBindingRevision: joined.rootBindingRevision.toString(), + projectRevision: joined.projectUpdatedAt.toISOString(), + workPackageId: linked.workPackageId, + agentRunId: linked.agentRunId, + taskAttemptId: linked.taskAttemptId, + policy: { + projectId: joined.projectId, + scopedProjectId: joined.projectId, + policyVersion, + allowedCapabilities: [ + ...(repositoryReadAllowed ? ['filesystem.project.read'] : []), + ], + policyCeilings: { + repository_read: repositoryReadAllowed, + }, + }, + } +} + +export function loadTrustedOperationContext( + input: TrustedOperationContextInput, +): Promise { + return loadTrustedOperationContextWithDependencies(input, productionContextDependencies) +} + +/** Dependency injection is exposed only for deterministic unit tests. */ +export function loadTrustedOperationContextForTest( + input: TrustedOperationContextInput, + dependencies: TrustedOperationContextLoaderDependencies, +): Promise { + return loadTrustedOperationContextWithDependencies(input, dependencies) +} + +/** Production entry point: authoritative context load followed by execution. */ +export async function executeTrustedOperation(input: { + request: unknown + taskId: string + attemptKey: string + workPackageId?: string | null + agentRunId?: string | null + taskAttemptId?: string | null +}) { + const { productionOperationAdapters } = await import('./production-adapters') + const context = await loadTrustedOperationContext({ + taskId: input.taskId, + attemptKey: input.attemptKey, + workPackageId: input.workPackageId, + agentRunId: input.agentRunId, + taskAttemptId: input.taskAttemptId, + }) + return executeOperation({ + request: input.request, + context, + dependencies: { + adapters: productionOperationAdapters, + ledger: databaseOperationLedger, + }, + }) +} diff --git a/web/worker/operations/executor.ts b/web/worker/operations/executor.ts new file mode 100644 index 00000000..6bac9391 --- /dev/null +++ b/web/worker/operations/executor.ts @@ -0,0 +1,624 @@ +import path from 'node:path' + +import type { ExecutionOutcome, ExecutionStopReasonCode } from '@/lib/execution-outcomes' +import { + OPERATION_PHASES, + isPlainRecord, + isSha256Fingerprint, + operationFingerprint, + parseOperationRequest, + type OperationAdapterKind, + type OperationExecutionResult, + type OperationPhase, + type OperationPolicyDecision, + type OperationRequest, + type OperationRunStatus, + type OperationVerificationStatus, +} from '@/lib/operations/contracts' +import { + OPERATION_CATALOG, + resolveOperationDefinition, + validateOperationInputs, +} from '@/lib/operations/catalog' +import { + evaluateOperationPolicy, + type TrustedOperationPolicyContext, +} from '@/lib/operations/policy' + +const UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i +const MAX_ATTEMPT_KEY_LENGTH = 200 +const MAX_PROJECT_ROOT_LENGTH = 4_096 +const ADAPTER_ABORT_SETTLEMENT_GRACE_MS = 250 + +export type TrustedOperationContext = { + taskId: string + projectId: string + attemptKey: string + projectRoot: string | null + rootBindingRevision: string + projectRevision: string + workPackageId?: string | null + agentRunId?: string | null + taskAttemptId?: string | null + policy: TrustedOperationPolicyContext +} + +export type OperationAdapterResult = { + output: unknown + evidenceRefs: string[] +} + +export type OperationAdapterExecution = { + signal: AbortSignal + deadline: Date +} + +export type FixedOperationAdapters = { + repositoryStatusRead: (context: TrustedOperationContext, execution: OperationAdapterExecution) => Promise + repositoryDiffSummary: (context: TrustedOperationContext, execution: OperationAdapterExecution) => Promise + repositoryBranchRead: (context: TrustedOperationContext, execution: OperationAdapterExecution) => Promise +} + +export type OperationRunStart = { + taskId: string + projectId: string + workPackageId: string | null + agentRunId: string | null + taskAttemptId: string | null + definitionSchemaVersion: 1 + operationId: string + operationVersion: number + capability: string + idempotencyKey: string + definitionDigest: string + scopeFingerprint: string + requestFingerprint: string + inputsFingerprint: string + reasonFingerprint: string + policyDecision: OperationPolicyDecision +} + +export type OperationRunEventWrite = { + runId: string + sequence: number + phase: OperationPhase + status: 'passed' | 'blocked' | 'failed' + detailCode: string + detailFingerprint: string + evidenceRefs: string[] +} + +export type OperationRunFinalization = { + runId: string + taskId: string + workPackageId: string | null + agentRunId: string | null + taskAttemptId: string | null + attemptKey: string + status: Exclude + verificationStatus: OperationVerificationStatus + outputFingerprint: string | null + policyDecision: OperationPolicyDecision + outcome: ExecutionOutcome + outcomeEvent: OperationRunEventWrite +} + +export type OperationLedger = { + begin(input: OperationRunStart): Promise< + | { kind: 'started'; runId: string } + | { + kind: 'replayed' + requestFingerprint: string + definitionDigest: string + scopeFingerprint: string + result: OperationExecutionResult + } + > + appendEvent(event: OperationRunEventWrite): Promise + /** Atomically stores the canonical outcome, its phase event, and run linkage. */ + finalize(input: OperationRunFinalization): Promise<{ executionOutcomeId: string }> +} + +export type OperationExecutorDependencies = { + adapters: FixedOperationAdapters + ledger: OperationLedger + catalog?: Parameters[1] +} + +export class OperationAdapterExecutionError extends Error { + readonly code: 'adapter_failed' | 'cancelled' | 'timeout' + readonly evidenceRefs: string[] + + constructor(code: OperationAdapterExecutionError['code'], evidenceRefs: string[] = []) { + super(code === 'timeout' + ? 'The deterministic operation timed out.' + : code === 'cancelled' + ? 'The deterministic operation was cancelled.' + : 'The deterministic operation adapter failed.') + this.name = 'OperationAdapterExecutionError' + this.code = code + this.evidenceRefs = normalizedEvidenceRefs(evidenceRefs) + } +} + +class OperationTimeoutError extends OperationAdapterExecutionError { + constructor(evidenceRefs: string[] = []) { + super('timeout', evidenceRefs) + this.name = 'OperationTimeoutError' + } +} + +class MissingRepositoryContextError extends Error { + constructor() { + super('The trusted project has no valid approved repository root.') + this.name = 'MissingRepositoryContextError' + } +} + +class OperationAdapterSettlementError extends Error { + constructor() { + super('Operation adapter did not settle after cancellation; the run requires recovery and was not terminalized.') + this.name = 'OperationAdapterSettlementError' + } +} + +function assertUuid(value: string | null | undefined, label: string, optional = false): void { + if (optional && value == null) return + if (typeof value !== 'string' || !UUID_PATTERN.test(value)) throw new Error(`${label} must be a UUID.`) +} + +function assertTrustedContext(context: TrustedOperationContext): void { + assertUuid(context.taskId, 'Task id') + assertUuid(context.projectId, 'Project id') + assertUuid(context.workPackageId, 'Work package id', true) + assertUuid(context.agentRunId, 'Agent run id', true) + assertUuid(context.taskAttemptId, 'Task attempt id', true) + assertUuid(context.policy.projectId, 'Policy project id') + assertUuid(context.policy.scopedProjectId, 'Policy scoped project id') + if (context.policy.projectId !== context.projectId) throw new Error('Policy context is not bound to the trusted project.') + if (!context.attemptKey || context.attemptKey.length > MAX_ATTEMPT_KEY_LENGTH || /[\u0000-\u001f\u007f]/u.test(context.attemptKey)) { + throw new Error('Attempt key must be bounded printable text.') + } + if (!/^\d{1,20}$/.test(context.rootBindingRevision)) { + throw new Error('Root binding revision must be a bounded decimal string.') + } + if (!context.projectRevision || context.projectRevision.length > 200 || /[\u0000-\u001f\u007f]/u.test(context.projectRevision)) { + throw new Error('Project revision must be bounded printable text.') + } + if (!context.policy.policyVersion || context.policy.policyVersion.length > 200 || /[\u0000-\u001f\u007f]/u.test(context.policy.policyVersion)) { + throw new Error('Policy version must be bounded printable text.') + } + if (context.policy.allowedCapabilities.some((capability) => typeof capability !== 'string' || capability.length < 1 || capability.length > 200)) { + throw new Error('Trusted capabilities must be non-empty bounded strings.') + } + if ( + typeof context.policy.policyCeilings.repository_read !== 'boolean' + ) { + throw new Error('Trusted operation policy ceilings must be explicit booleans.') + } +} + +function assertEvidenceRefs(value: unknown): asserts value is string[] { + if (!Array.isArray(value) || value.length > 100 || value.some((ref) => typeof ref !== 'string' || !UUID_PATTERN.test(ref))) { + throw new Error('Operation evidence references must be a bounded array of UUIDs.') + } +} + +function normalizedEvidenceRefs(refs: string[]): string[] { + assertEvidenceRefs(refs) + return [...new Set(refs.map((ref) => ref.toLowerCase()))].sort() +} + +function adapterFor(kind: OperationAdapterKind, adapters: FixedOperationAdapters): FixedOperationAdapters[keyof FixedOperationAdapters] { + switch (kind) { + case 'repository_status_read_v1': return adapters.repositoryStatusRead + case 'repository_diff_summary_v1': return adapters.repositoryDiffSummary + case 'repository_branch_read_v1': return adapters.repositoryBranchRead + } +} + +function validatePreflight(kind: OperationAdapterKind, context: TrustedOperationContext): void { + if ( + context.projectRoot === null + || !context.projectRoot + || context.projectRoot.length > MAX_PROJECT_ROOT_LENGTH + || context.projectRoot.includes('\0') + || !path.isAbsolute(context.projectRoot) + ) { + throw new MissingRepositoryContextError() + } +} + +function verifyAdapterOutput(kind: OperationAdapterKind, output: unknown, evidenceRefs: string[]): boolean { + if (!isPlainRecord(output)) return false + if ( + kind === 'repository_status_read_v1' + || kind === 'repository_diff_summary_v1' + || kind === 'repository_branch_read_v1' + ) { + return output.exitCode === 0 && typeof output.summary === 'string' && evidenceRefs.length > 0 + } + return false +} + +async function runAdapterWithTimeout( + adapter: FixedOperationAdapters[keyof FixedOperationAdapters], + context: TrustedOperationContext, + timeoutMs: number, +): Promise { + const controller = new AbortController() + const deadline = new Date(Date.now() + timeoutMs) + type AdapterSettlement = + | { kind: 'completed'; value: OperationAdapterResult } + | { kind: 'failed'; error: unknown } + const settlement: Promise = adapter(context, { + signal: controller.signal, + deadline, + }).then( + (value) => ({ kind: 'completed' as const, value }), + (error: unknown) => ({ kind: 'failed' as const, error }), + ) + let timer: ReturnType | undefined + try { + const first = await Promise.race([ + settlement, + new Promise<{ kind: 'timeout' }>((resolve) => { + timer = setTimeout(() => { + controller.abort() + resolve({ kind: 'timeout' }) + }, timeoutMs) + }), + ]) + if (first.kind === 'timeout') { + // Built-in adapters are cancellation-aware. Do not terminalize the run + // until the adapter has stopped, including its audit/evidence work. + let graceTimer: ReturnType | undefined + const stopped = await Promise.race([ + settlement, + new Promise<{ kind: 'pending' }>((resolve) => { + graceTimer = setTimeout(() => resolve({ kind: 'pending' }), ADAPTER_ABORT_SETTLEMENT_GRACE_MS) + }), + ]) + if (graceTimer) clearTimeout(graceTimer) + if (stopped.kind === 'pending') throw new OperationAdapterSettlementError() + const evidenceRefs = stopped.kind === 'completed' + ? stopped.value.evidenceRefs + : stopped.error instanceof OperationAdapterExecutionError + ? stopped.error.evidenceRefs + : [] + throw new OperationTimeoutError(evidenceRefs) + } + if (first.kind === 'failed') throw first.error + return first.value + } finally { + if (timer) clearTimeout(timer) + } +} + +function phaseEvent(input: Omit & { detail: unknown }): OperationRunEventWrite { + const sequence = OPERATION_PHASES.indexOf(input.phase) + if (sequence < 0) throw new Error('Unknown operation phase.') + const detailFingerprint = operationFingerprint(`event:${input.phase}`, input.detail) + if (!isSha256Fingerprint(detailFingerprint)) throw new Error('Invalid operation event fingerprint.') + return { + runId: input.runId, + sequence, + phase: input.phase, + status: input.status, + detailCode: input.detailCode, + detailFingerprint, + evidenceRefs: normalizedEvidenceRefs(input.evidenceRefs), + } +} + +function canonicalOutcome(input: { + result: 'completed' | 'blocked' | 'failed' + code: ExecutionStopReasonCode | null + summary: string | null + evidenceRefs?: string[] +}): ExecutionOutcome { + return { + schemaVersion: 1, + transportStatus: 'ok', + result: input.result, + stopReasonCode: input.code, + stopReasonSummary: input.summary, + retryable: false, + evidenceRefs: normalizedEvidenceRefs(input.evidenceRefs ?? []), + verifierRequired: false, + verificationStatus: 'not_required', + } +} + +function policyOutcome(decision: OperationPolicyDecision): ExecutionOutcome { + const missingCapability = decision.code === 'missing_capability' || decision.code === 'capability_mismatch' + return canonicalOutcome({ + result: 'blocked', + code: missingCapability ? 'missing_capability' : 'policy_blocked', + summary: `Operation policy denied the request (${decision.code}).`, + }) +} + +async function terminalize(input: { + ledger: OperationLedger + runId: string + context: TrustedOperationContext + definitionDigest: string + scopeFingerprint: string + request: OperationRequest + status: Exclude + verificationStatus: OperationVerificationStatus + output: unknown | null + policyDecision: OperationPolicyDecision + outcome: ExecutionOutcome +}): Promise { + const outputFingerprint = input.output === null ? null : operationFingerprint('output', input.output) + const outcomeEvent = phaseEvent({ + runId: input.runId, + phase: 'outcome', + status: input.status === 'completed' ? 'passed' : input.status, + detailCode: input.outcome.stopReasonCode ?? 'completed', + detail: input.outcome, + evidenceRefs: input.outcome.evidenceRefs, + }) + await input.ledger.finalize({ + runId: input.runId, + taskId: input.context.taskId, + workPackageId: input.context.workPackageId ?? null, + agentRunId: input.context.agentRunId ?? null, + taskAttemptId: input.context.taskAttemptId ?? null, + attemptKey: `operation:${operationFingerprint('outcome-attempt', { + taskId: input.context.taskId, + attemptKey: input.context.attemptKey, + operationId: input.request.operationId, + operationVersion: input.request.operationVersion, + })}`, + status: input.status, + verificationStatus: input.verificationStatus, + outputFingerprint, + policyDecision: input.policyDecision, + outcome: input.outcome, + outcomeEvent, + }) + return { + runId: input.runId, + operationId: input.request.operationId, + operationVersion: input.request.operationVersion, + definitionDigest: input.definitionDigest, + scopeFingerprint: input.scopeFingerprint, + status: input.status, + verificationStatus: input.verificationStatus, + replayed: false, + output: input.output, + outcome: input.outcome, + } +} + +export async function executeOperation(input: { + request: unknown + context: TrustedOperationContext + dependencies: OperationExecutorDependencies +}): Promise { + assertTrustedContext(input.context) + const request = parseOperationRequest(input.request) + const definition = resolveOperationDefinition(request, input.dependencies.catalog ?? OPERATION_CATALOG) + validateOperationInputs(definition, request.inputs) + + const definitionDigest = operationFingerprint('definition', definition) + const scopeFingerprint = operationFingerprint('scope', { + projectId: input.context.projectId, + scopedProjectId: input.context.policy.scopedProjectId, + projectRoot: input.context.projectRoot, + rootBindingRevision: input.context.rootBindingRevision, + projectRevision: input.context.projectRevision, + policyVersion: input.context.policy.policyVersion, + allowedCapabilities: [...new Set(input.context.policy.allowedCapabilities)].sort(), + policyCeilings: { + repository_read: input.context.policy.policyCeilings.repository_read, + }, + }) + // `reason` is audit-only: it is fingerprinted separately and cannot affect + // idempotency, adapter selection, execution, verification, or replay. + const requestFingerprint = operationFingerprint('request', { + schemaVersion: request.schemaVersion, + operationId: request.operationId, + operationVersion: request.operationVersion, + inputs: request.inputs, + requestedCapability: request.requestedCapability, + }) + const policyDecision = evaluateOperationPolicy({ + definition, + requestedCapability: request.requestedCapability, + context: input.context.policy, + }) + const run = await input.dependencies.ledger.begin({ + taskId: input.context.taskId, + projectId: input.context.projectId, + workPackageId: input.context.workPackageId ?? null, + agentRunId: input.context.agentRunId ?? null, + taskAttemptId: input.context.taskAttemptId ?? null, + definitionSchemaVersion: 1, + operationId: definition.id, + operationVersion: definition.version, + capability: definition.capability, + idempotencyKey: operationFingerprint('idempotency', { + taskId: input.context.taskId, + attemptKey: input.context.attemptKey, + operationId: request.operationId, + operationVersion: request.operationVersion, + }), + definitionDigest, + scopeFingerprint, + requestFingerprint, + inputsFingerprint: operationFingerprint('inputs', request.inputs), + reasonFingerprint: operationFingerprint('reason', request.reason), + policyDecision, + }) + + if (run.kind === 'replayed') { + if (run.requestFingerprint !== requestFingerprint || run.definitionDigest !== definitionDigest || run.scopeFingerprint !== scopeFingerprint) { + throw new Error('Idempotency key is already bound to a different operation request or scope.') + } + if (run.result.status === 'running') { + throw new Error('Operation attempt is still in progress or requires recovery; use a new attempt key after auditing the incomplete run.') + } + return { ...run.result, replayed: true, output: null } + } + + await input.dependencies.ledger.appendEvent(phaseEvent({ + runId: run.runId, + phase: 'request_validation', + status: 'passed', + detailCode: 'request_valid', + detail: { requestFingerprint }, + evidenceRefs: [], + })) + + await input.dependencies.ledger.appendEvent(phaseEvent({ + runId: run.runId, + phase: 'policy', + status: policyDecision.allowed ? 'passed' : 'blocked', + detailCode: policyDecision.code, + detail: policyDecision, + evidenceRefs: [], + })) + if (!policyDecision.allowed) { + return terminalize({ + ledger: input.dependencies.ledger, + runId: run.runId, + context: input.context, + definitionDigest, + scopeFingerprint, + request, + status: 'blocked', + verificationStatus: 'not_started', + output: null, + policyDecision, + outcome: policyOutcome(policyDecision), + }) + } + + const failPhase = async ( + phase: 'preflight' | 'execution', + error: unknown, + ): Promise => { + const timeout = error instanceof OperationTimeoutError + const missingRepositoryContext = error instanceof MissingRepositoryContextError + const failureEvidenceRefs = error instanceof OperationAdapterExecutionError + ? error.evidenceRefs + : [] + await input.dependencies.ledger.appendEvent(phaseEvent({ + runId: run.runId, + phase, + status: 'failed', + detailCode: timeout ? 'timeout' : phase === 'preflight' ? 'preflight_failed' : 'adapter_failed', + detail: { code: timeout ? 'timeout' : missingRepositoryContext ? 'missing_repository_context' : phase === 'preflight' ? 'preflight_failed' : 'adapter_failed' }, + evidenceRefs: failureEvidenceRefs, + })) + return terminalize({ + ledger: input.dependencies.ledger, + runId: run.runId, + context: input.context, + definitionDigest, + scopeFingerprint, + request, + status: 'failed', + verificationStatus: 'not_started', + output: null, + policyDecision, + outcome: canonicalOutcome({ + result: 'failed', + code: timeout ? 'timeout' : missingRepositoryContext ? 'missing_repository_context' : 'unknown', + evidenceRefs: failureEvidenceRefs, + summary: timeout + ? 'The deterministic operation timed out.' + : missingRepositoryContext + ? 'The trusted project has no valid approved repository root.' + : `The deterministic operation ${phase} failed.`, + }), + }) + } + + try { + validatePreflight(definition.adapter, input.context) + } catch (error) { + return failPhase('preflight', error) + } + await input.dependencies.ledger.appendEvent(phaseEvent({ + runId: run.runId, + phase: 'preflight', + status: 'passed', + detailCode: 'scope_ready', + detail: { scopeFingerprint }, + evidenceRefs: [], + })) + + let adapterResult: OperationAdapterResult + let evidenceRefs: string[] + let outputFingerprint: string + try { + adapterResult = await runAdapterWithTimeout( + adapterFor(definition.adapter, input.dependencies.adapters), + input.context, + definition.timeoutMs, + ) + assertEvidenceRefs(adapterResult.evidenceRefs) + evidenceRefs = normalizedEvidenceRefs(adapterResult.evidenceRefs) + outputFingerprint = operationFingerprint('output', adapterResult.output) + } catch (error) { + if (error instanceof OperationAdapterSettlementError) throw error + return failPhase('execution', error) + } + await input.dependencies.ledger.appendEvent(phaseEvent({ + runId: run.runId, + phase: 'execution', + status: 'passed', + detailCode: 'adapter_completed', + detail: { outputFingerprint }, + evidenceRefs, + })) + + const verified = verifyAdapterOutput(definition.adapter, adapterResult.output, evidenceRefs) + await input.dependencies.ledger.appendEvent(phaseEvent({ + runId: run.runId, + phase: 'verification', + status: verified ? 'passed' : 'failed', + detailCode: verified ? 'deterministic_verification_passed' : 'deterministic_verification_failed', + detail: { adapter: definition.adapter, outputFingerprint }, + evidenceRefs, + })) + if (!verified) { + return terminalize({ + ledger: input.dependencies.ledger, + runId: run.runId, + context: input.context, + definitionDigest, + scopeFingerprint, + request, + status: 'failed', + verificationStatus: 'failed', + output: adapterResult.output, + policyDecision, + outcome: canonicalOutcome({ + result: 'failed', + code: 'validation_failed', + summary: 'Deterministic operation verification failed.', + evidenceRefs, + }), + }) + } + + return terminalize({ + ledger: input.dependencies.ledger, + runId: run.runId, + context: input.context, + definitionDigest, + scopeFingerprint, + request, + status: 'completed', + verificationStatus: 'passed', + output: adapterResult.output, + policyDecision, + outcome: canonicalOutcome({ result: 'completed', code: null, summary: null, evidenceRefs }), + }) +} diff --git a/web/worker/operations/ledger.ts b/web/worker/operations/ledger.ts new file mode 100644 index 00000000..5ac274fa --- /dev/null +++ b/web/worker/operations/ledger.ts @@ -0,0 +1,212 @@ +import { and, eq } from 'drizzle-orm' + +import { db } from '@/db' +import { executionOutcomes, operationRunEvents, operationRuns } from '@/db/schema' +import { + isExecutionOutcome, + normalizeExecutionOutcome, + type ExecutionOutcome, +} from '@/lib/execution-outcomes' +import { + operationFingerprint, + type OperationExecutionResult, + type OperationRunStatus, + type OperationVerificationStatus, +} from '@/lib/operations/contracts' +import { sanitizeWorkerMessage } from '@/worker/redaction' +import type { + OperationLedger, + OperationRunEventWrite, + OperationRunFinalization, + OperationRunStart, +} from './executor' + +type LedgerTransaction = Parameters[0]>[0] + +function runStatus(value: string): OperationRunStatus { + if (value === 'running' || value === 'completed' || value === 'blocked' || value === 'failed') return value + throw new Error('Stored operation run has an invalid status.') +} + +function verificationStatus(value: string): OperationVerificationStatus { + if (value === 'not_started' || value === 'passed' || value === 'failed') return value + throw new Error('Stored operation run has an invalid verification status.') +} + +function storedOutcome(row: typeof executionOutcomes.$inferSelect): ExecutionOutcome { + const outcome: unknown = { + schemaVersion: row.schemaVersion, + transportStatus: row.transportStatus, + result: row.result, + stopReasonCode: row.stopReasonCode, + stopReasonSummary: row.stopReasonSummary, + retryable: row.retryable, + evidenceRefs: row.evidenceRefs, + verifierRequired: row.verifierRequired, + verificationStatus: row.verificationStatus, + } + if (!isExecutionOutcome(outcome)) throw new Error('Stored operation outcome is invalid.') + return outcome +} + +async function replayResult( + tx: LedgerTransaction, + row: typeof operationRuns.$inferSelect, +): Promise { + let outcome: ExecutionOutcome | null = null + if (row.executionOutcomeId) { + const [stored] = await tx + .select() + .from(executionOutcomes) + .where(eq(executionOutcomes.id, row.executionOutcomeId)) + .limit(1) + if (!stored) throw new Error('Operation run references a missing canonical outcome.') + outcome = storedOutcome(stored) + if (row.outcomeFingerprint !== operationFingerprint('canonical-outcome', outcome)) { + throw new Error('Stored canonical operation outcome does not match its immutable fingerprint.') + } + } else if (row.status !== 'running') { + throw new Error('Terminal operation run has no canonical outcome linkage.') + } + return { + runId: row.id, + operationId: row.operationId, + operationVersion: row.operationVersion, + definitionDigest: row.definitionDigest, + scopeFingerprint: row.scopeFingerprint, + status: runStatus(row.status), + verificationStatus: verificationStatus(row.verificationStatus), + replayed: true, + output: null, + outcome, + } +} + +function eventValues(event: OperationRunEventWrite) { + return { + operationRunId: event.runId, + sequence: event.sequence, + phase: event.phase, + status: event.status, + detailCode: event.detailCode, + detailFingerprint: event.detailFingerprint, + evidenceRefs: event.evidenceRefs, + } +} + +export function createDatabaseOperationLedger(database: typeof db = db): OperationLedger { + return { + async begin(input: OperationRunStart) { + return database.transaction(async (tx) => { + const [inserted] = await tx + .insert(operationRuns) + .values({ + taskId: input.taskId, + projectId: input.projectId, + workPackageId: input.workPackageId, + agentRunId: input.agentRunId, + taskAttemptId: input.taskAttemptId, + definitionSchemaVersion: input.definitionSchemaVersion, + operationId: input.operationId, + operationVersion: input.operationVersion, + capability: input.capability, + idempotencyKey: input.idempotencyKey, + definitionDigest: input.definitionDigest, + scopeFingerprint: input.scopeFingerprint, + requestFingerprint: input.requestFingerprint, + inputsFingerprint: input.inputsFingerprint, + reasonFingerprint: input.reasonFingerprint, + policyDecision: { ...input.policyDecision }, + }) + .onConflictDoNothing({ + target: [operationRuns.taskId, operationRuns.idempotencyKey], + }) + .returning({ id: operationRuns.id }) + if (inserted) return { kind: 'started' as const, runId: inserted.id } + + const [existing] = await tx + .select() + .from(operationRuns) + .where(and( + eq(operationRuns.taskId, input.taskId), + eq(operationRuns.idempotencyKey, input.idempotencyKey), + )) + .limit(1) + if (!existing) throw new Error('Idempotent operation run could not be resolved.') + return { + kind: 'replayed' as const, + requestFingerprint: existing.requestFingerprint, + definitionDigest: existing.definitionDigest, + scopeFingerprint: existing.scopeFingerprint, + result: await replayResult(tx, existing), + } + }) + }, + + async appendEvent(event: OperationRunEventWrite) { + await database.insert(operationRunEvents).values(eventValues(event)) + }, + + async finalize(input: OperationRunFinalization) { + return database.transaction(async (tx) => { + const outcome = normalizeExecutionOutcome(input.outcome, sanitizeWorkerMessage) + const [outcomeRow] = await tx + .insert(executionOutcomes) + .values({ + taskId: input.taskId, + workPackageId: input.workPackageId, + agentRunId: input.agentRunId, + taskAttemptId: input.taskAttemptId, + attemptKey: input.attemptKey, + schemaVersion: outcome.schemaVersion, + transportStatus: outcome.transportStatus, + result: outcome.result, + stopReasonCode: outcome.stopReasonCode, + stopReasonSummary: outcome.stopReasonSummary, + retryable: outcome.retryable, + evidenceRefs: outcome.evidenceRefs, + verifierRequired: outcome.verifierRequired, + verificationStatus: outcome.verificationStatus, + updatedAt: new Date(), + }) + .onConflictDoUpdate({ + target: [executionOutcomes.taskId, executionOutcomes.attemptKey], + set: { + workPackageId: input.workPackageId, + agentRunId: input.agentRunId, + taskAttemptId: input.taskAttemptId, + transportStatus: outcome.transportStatus, + result: outcome.result, + stopReasonCode: outcome.stopReasonCode, + stopReasonSummary: outcome.stopReasonSummary, + retryable: outcome.retryable, + evidenceRefs: outcome.evidenceRefs, + verifierRequired: outcome.verifierRequired, + verificationStatus: outcome.verificationStatus, + updatedAt: new Date(), + }, + }) + .returning({ id: executionOutcomes.id }) + if (!outcomeRow) throw new Error('Canonical operation outcome was not stored.') + + await tx.insert(operationRunEvents).values(eventValues(input.outcomeEvent)) + const [completed] = await tx + .update(operationRuns) + .set({ + executionOutcomeId: outcomeRow.id, + status: input.status, + verificationStatus: input.verificationStatus, + outputFingerprint: input.outputFingerprint, + outcomeFingerprint: operationFingerprint('canonical-outcome', outcome), + completedAt: new Date(), + }) + .where(and(eq(operationRuns.id, input.runId), eq(operationRuns.status, 'running'))) + .returning({ id: operationRuns.id }) + if (!completed) throw new Error('Operation run was not in a finalizable state.') + return { executionOutcomeId: outcomeRow.id } + }) + }, + } +} + +export const databaseOperationLedger = createDatabaseOperationLedger() diff --git a/web/worker/operations/production-adapters.ts b/web/worker/operations/production-adapters.ts new file mode 100644 index 00000000..96d8a4cb --- /dev/null +++ b/web/worker/operations/production-adapters.ts @@ -0,0 +1,112 @@ +import { eq } from 'drizzle-orm' + +import { db } from '@/db' +import { projects, type Project } from '@/db/schema' +import { assertProjectLocalPathForExecution } from '@/lib/projects/local-path' +import { + recordScopedCommandAudit, + runScopedRepositoryCommand, +} from '@/worker/repository-evidence' +import { + OperationAdapterExecutionError, + type FixedOperationAdapters, + type OperationAdapterExecution, + type OperationAdapterResult, + type TrustedOperationContext, +} from './executor' +import { loadTrustedOperationContext } from './context' + +function assertNotAborted(execution: OperationAdapterExecution): void { + if (execution.signal.aborted || Date.now() >= execution.deadline.getTime()) { + throw new Error('Deterministic operation was aborted.') + } +} + +async function currentProject(context: TrustedOperationContext): Promise { + const [project] = await db.select().from(projects).where(eq(projects.id, context.projectId)).limit(1) + if (!project) throw new Error('The trusted operation project no longer exists.') + if ( + project.rootBindingRevision.toString() !== context.rootBindingRevision + || project.updatedAt.toISOString() !== context.projectRevision + ) { + throw new Error('The trusted operation project scope changed before adapter execution.') + } + return project +} + +async function assertCurrentAuthority(context: TrustedOperationContext): Promise { + const fresh = await loadTrustedOperationContext({ + taskId: context.taskId, + attemptKey: context.attemptKey, + workPackageId: context.workPackageId, + agentRunId: context.agentRunId, + taskAttemptId: context.taskAttemptId, + }) + if ( + fresh.projectId !== context.projectId + || fresh.projectRoot !== context.projectRoot + || fresh.rootBindingRevision !== context.rootBindingRevision + || fresh.projectRevision !== context.projectRevision + || fresh.policy.policyVersion !== context.policy.policyVersion + || fresh.policy.policyCeilings.repository_read !== true + || !fresh.policy.allowedCapabilities.includes('filesystem.project.read') + ) { + throw new Error('The authoritative operation policy changed before repository execution.') + } +} + +async function repositoryRead( + context: TrustedOperationContext, + execution: OperationAdapterExecution, + argv: string[], +): Promise { + assertNotAborted(execution) + await assertCurrentAuthority(context) + assertNotAborted(execution) + const project = await currentProject(context) + assertNotAborted(execution) + if (!context.projectRoot) throw new Error('The trusted project has no approved repository root.') + const currentRoot = await assertProjectLocalPathForExecution(project) + assertNotAborted(execution) + if (currentRoot !== context.projectRoot) { + throw new Error('The canonical project root changed before repository execution.') + } + const result = await runScopedRepositoryCommand({ + cwd: context.projectRoot, + command: 'git', + argv, + signal: execution.signal, + }) + const audit = await recordScopedCommandAudit({ + result, + taskId: context.taskId, + workPackageId: context.workPackageId, + agentRunId: context.agentRunId, + }) + if (execution.signal.aborted || Date.now() >= execution.deadline.getTime()) { + throw new OperationAdapterExecutionError('cancelled', [audit.id]) + } + if (result.exitCode !== 0) throw new OperationAdapterExecutionError('adapter_failed', [audit.id]) + return { + output: { exitCode: result.exitCode, summary: result.outputSummary }, + evidenceRefs: [audit.id], + } +} + +export const productionOperationAdapters: FixedOperationAdapters = { + repositoryStatusRead: (context, execution) => repositoryRead( + context, + execution, + ['status', '--short'], + ), + repositoryDiffSummary: (context, execution) => repositoryRead( + context, + execution, + ['diff', '--no-ext-diff', '--no-textconv', '--stat', '--'], + ), + repositoryBranchRead: (context, execution) => repositoryRead( + context, + execution, + ['branch', '--show-current'], + ), +} diff --git a/web/worker/repository-evidence.ts b/web/worker/repository-evidence.ts index 8d08d09a..29fa5493 100644 --- a/web/worker/repository-evidence.ts +++ b/web/worker/repository-evidence.ts @@ -89,10 +89,11 @@ export type ScopedCommandAuditSink = ( record: ScopedCommandAuditRecord, ) => Promise<{ id: string }> -type ScopedCommandInput = { +export type ScopedCommandInput = { cwd: string command: string argv: string[] + signal?: AbortSignal } function truncate(value: string, maxBytes = MAX_OUTPUT_BYTES): string { @@ -387,10 +388,10 @@ function isAllowedGitReadOnly(argv: string[]): boolean { 'status --short', 'branch --show-current', 'remote -v', - 'diff --stat', - 'diff --stat --', - 'diff --stat HEAD --', - 'diff --name-status', + 'diff --no-ext-diff --no-textconv --stat', + 'diff --no-ext-diff --no-textconv --stat --', + 'diff --no-ext-diff --no-textconv --stat HEAD --', + 'diff --no-ext-diff --no-textconv --name-status', ].includes(normalized) } @@ -494,6 +495,7 @@ export async function runScopedRepositoryCommand(input: ScopedCommandInput): Pro env: scopedCommandEnv(), maxBuffer: Math.max(MAX_OUTPUT_BYTES, MAX_DIFF_BYTES) * 2, timeout: COMMAND_TIMEOUT_MS, + signal: input.signal, }) const stdout = truncate(result.stdout, input.command === 'git' && input.argv[0] === 'diff' ? MAX_DIFF_BYTES : MAX_OUTPUT_BYTES) const stderr = truncate(result.stderr) diff --git a/web/worker/work-package-handoff.ts b/web/worker/work-package-handoff.ts index a06c1038..310b5a7a 100644 --- a/web/worker/work-package-handoff.ts +++ b/web/worker/work-package-handoff.ts @@ -3129,7 +3129,7 @@ async function executeReadyWorkPackage( const diffResult = await runScopedRepositoryCommand({ cwd: repositoryContext.projectLocalPath, command: 'git', - argv: ['diff', '--stat', 'HEAD', '--'], + argv: ['diff', '--no-ext-diff', '--no-textconv', '--stat', 'HEAD', '--'], }) assertQueueClaimOwned(options) diffSummary = diffResult.outputSummary || 'No tracked-file diff detected.' @@ -3140,7 +3140,7 @@ async function executeReadyWorkPackage( executionLease: { runId: run.id }, metadata: { artifactKind: 'repository_diff_summary', - command: ['git', 'diff', '--stat', 'HEAD', '--'], + command: ['git', 'diff', '--no-ext-diff', '--no-textconv', '--stat', 'HEAD', '--'], exitCode: diffResult.exitCode, riskClass: diffResult.riskClass, source: 'repository-evidence', From 4b2017d2398c8436c49d2d8de9fb6757f83d63fb Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 02:37:05 +0000 Subject: [PATCH 2/3] fix: revoke default PUBLIC execute on the new operation-ledger trigger functions New PL/pgSQL functions default to PUBLIC EXECUTE in Postgres unless explicitly revoked, matching every other custom routine in this codebase (e.g. migration 0025's REVOKE ALL ... FROM PUBLIC for forge_epic_172_reject_mutation_v1, and the block of REVOKEs in migration 0027). Migration 0030 skipped this for its three new trigger functions, so forge_project_root_reconciler picked up implicit EXECUTE via PUBLIC and failed the closed root-reconciler effective-privilege allowlist proof in CI ("unexpected": [...forge_guard_operation_event_insert_v1, ...]). --- web/db/migrations/0030_operation_runs.sql | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/web/db/migrations/0030_operation_runs.sql b/web/db/migrations/0030_operation_runs.sql index 0cb486eb..d8fec0a2 100644 --- a/web/db/migrations/0030_operation_runs.sql +++ b/web/db/migrations/0030_operation_runs.sql @@ -131,6 +131,8 @@ BEGIN END; $$; --> statement-breakpoint +REVOKE ALL ON FUNCTION public.forge_guard_operation_event_insert_v1() FROM PUBLIC; +--> statement-breakpoint CREATE TRIGGER "operation_run_events_order_guard" BEFORE INSERT ON "operation_run_events" FOR EACH ROW EXECUTE FUNCTION "forge_guard_operation_event_insert_v1"(); @@ -172,6 +174,8 @@ BEGIN END; $$; --> statement-breakpoint +REVOKE ALL ON FUNCTION public.forge_guard_operation_run_history_v1() FROM PUBLIC; +--> statement-breakpoint CREATE TRIGGER "operation_runs_history_guard" BEFORE UPDATE OR DELETE ON "operation_runs" FOR EACH ROW EXECUTE FUNCTION "forge_guard_operation_run_history_v1"(); @@ -185,6 +189,8 @@ BEGIN END; $$; --> statement-breakpoint +REVOKE ALL ON FUNCTION public.forge_reject_operation_event_mutation_v1() FROM PUBLIC; +--> statement-breakpoint CREATE TRIGGER "operation_run_events_append_only" BEFORE UPDATE OR DELETE ON "operation_run_events" FOR EACH ROW EXECUTE FUNCTION "forge_reject_operation_event_mutation_v1"(); From f78758753736d925690a874a620095ce669284c5 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 02:58:49 +0000 Subject: [PATCH 3/3] fix: surface the real database error from operation-ledger writes drizzle-orm (^0.45.2) wraps every failed query in a DrizzleQueryError whose message is a generic "Failed query: insert into ... params: ..." dump; the actual driver/database error -- e.g. the text a trigger raises via RAISE EXCEPTION, such as "operation run events must be appended in phase order" -- is only reachable via `.cause`. appendEvent and finalize now unwrap it, so callers (and operators reading logs) see the real rejection reason instead of an opaque wrapper. Caught by operation-ledger.postgres.test.ts's live-Postgres proof, which asserts on that exact trigger message. --- web/worker/operations/ledger.ts | 110 +++++++++++++++++++------------- 1 file changed, 65 insertions(+), 45 deletions(-) diff --git a/web/worker/operations/ledger.ts b/web/worker/operations/ledger.ts index 5ac274fa..d30a4a08 100644 --- a/web/worker/operations/ledger.ts +++ b/web/worker/operations/ledger.ts @@ -82,6 +82,18 @@ async function replayResult( } } +/** + * drizzle-orm wraps every failed query in a DrizzleQueryError whose message + * is a generic "Failed query: ..." dump of the SQL and params; the real + * driver/database error (e.g. a trigger's RAISE EXCEPTION text) is only + * reachable via `.cause`. Surface that instead so callers -- and anyone + * reading logs -- see the actual reason a ledger write was rejected. + */ +function unwrapDatabaseError(err: unknown): never { + if (err instanceof Error && err.cause instanceof Error) throw err.cause + throw err +} + function eventValues(event: OperationRunEventWrite) { return { operationRunId: event.runId, @@ -144,37 +156,26 @@ export function createDatabaseOperationLedger(database: typeof db = db): Operati }, async appendEvent(event: OperationRunEventWrite) { - await database.insert(operationRunEvents).values(eventValues(event)) + try { + await database.insert(operationRunEvents).values(eventValues(event)) + } catch (err) { + unwrapDatabaseError(err) + } }, async finalize(input: OperationRunFinalization) { - return database.transaction(async (tx) => { - const outcome = normalizeExecutionOutcome(input.outcome, sanitizeWorkerMessage) - const [outcomeRow] = await tx - .insert(executionOutcomes) - .values({ - taskId: input.taskId, - workPackageId: input.workPackageId, - agentRunId: input.agentRunId, - taskAttemptId: input.taskAttemptId, - attemptKey: input.attemptKey, - schemaVersion: outcome.schemaVersion, - transportStatus: outcome.transportStatus, - result: outcome.result, - stopReasonCode: outcome.stopReasonCode, - stopReasonSummary: outcome.stopReasonSummary, - retryable: outcome.retryable, - evidenceRefs: outcome.evidenceRefs, - verifierRequired: outcome.verifierRequired, - verificationStatus: outcome.verificationStatus, - updatedAt: new Date(), - }) - .onConflictDoUpdate({ - target: [executionOutcomes.taskId, executionOutcomes.attemptKey], - set: { + try { + return await database.transaction(async (tx) => { + const outcome = normalizeExecutionOutcome(input.outcome, sanitizeWorkerMessage) + const [outcomeRow] = await tx + .insert(executionOutcomes) + .values({ + taskId: input.taskId, workPackageId: input.workPackageId, agentRunId: input.agentRunId, taskAttemptId: input.taskAttemptId, + attemptKey: input.attemptKey, + schemaVersion: outcome.schemaVersion, transportStatus: outcome.transportStatus, result: outcome.result, stopReasonCode: outcome.stopReasonCode, @@ -184,27 +185,46 @@ export function createDatabaseOperationLedger(database: typeof db = db): Operati verifierRequired: outcome.verifierRequired, verificationStatus: outcome.verificationStatus, updatedAt: new Date(), - }, - }) - .returning({ id: executionOutcomes.id }) - if (!outcomeRow) throw new Error('Canonical operation outcome was not stored.') + }) + .onConflictDoUpdate({ + target: [executionOutcomes.taskId, executionOutcomes.attemptKey], + set: { + workPackageId: input.workPackageId, + agentRunId: input.agentRunId, + taskAttemptId: input.taskAttemptId, + transportStatus: outcome.transportStatus, + result: outcome.result, + stopReasonCode: outcome.stopReasonCode, + stopReasonSummary: outcome.stopReasonSummary, + retryable: outcome.retryable, + evidenceRefs: outcome.evidenceRefs, + verifierRequired: outcome.verifierRequired, + verificationStatus: outcome.verificationStatus, + updatedAt: new Date(), + }, + }) + .returning({ id: executionOutcomes.id }) + if (!outcomeRow) throw new Error('Canonical operation outcome was not stored.') - await tx.insert(operationRunEvents).values(eventValues(input.outcomeEvent)) - const [completed] = await tx - .update(operationRuns) - .set({ - executionOutcomeId: outcomeRow.id, - status: input.status, - verificationStatus: input.verificationStatus, - outputFingerprint: input.outputFingerprint, - outcomeFingerprint: operationFingerprint('canonical-outcome', outcome), - completedAt: new Date(), - }) - .where(and(eq(operationRuns.id, input.runId), eq(operationRuns.status, 'running'))) - .returning({ id: operationRuns.id }) - if (!completed) throw new Error('Operation run was not in a finalizable state.') - return { executionOutcomeId: outcomeRow.id } - }) + await tx.insert(operationRunEvents).values(eventValues(input.outcomeEvent)) + const [completed] = await tx + .update(operationRuns) + .set({ + executionOutcomeId: outcomeRow.id, + status: input.status, + verificationStatus: input.verificationStatus, + outputFingerprint: input.outputFingerprint, + outcomeFingerprint: operationFingerprint('canonical-outcome', outcome), + completedAt: new Date(), + }) + .where(and(eq(operationRuns.id, input.runId), eq(operationRuns.status, 'running'))) + .returning({ id: operationRuns.id }) + if (!completed) throw new Error('Operation run was not in a finalizable state.') + return { executionOutcomeId: outcomeRow.id } + }) + } catch (err) { + unwrapDatabaseError(err) + } }, } }