From d53b7902aba0345880a53234d236dff5d769c5bf Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 9 Aug 2026 06:46:45 +0000 Subject: [PATCH] fix(native-runner): prove live Phase 6 effects Route corpus operations through cancellation, reconciliation, attention, and rollout entrypoints, materialize effects through owning services, and document the live-versus-policy proof boundary. Co-Authored-By: Paperclip --- .../design/phase-6-thin-paperclip-adapter.md | 40 ++ .../docs/phase-06-thin-paperclip-adapter.md | 22 +- .../phase-06-thin-paperclip-adapter.md | 29 +- .../2026-08-09-phase-06-verification.md | 38 +- .../2026-08-09-phase-06-implementation.md | 16 +- .../native-status-arbiter-corpus.test.ts | 491 ++++++++++++++++-- server/src/services/heartbeat.ts | 12 +- .../native-finalization-reconciler.ts | 204 +++++++- .../native-interaction-bridge.ts | 140 ++++- .../native-runtime/native-session-executor.ts | 134 ++++- .../native-workspace-finalizer.ts | 133 +++++ .../services/native-runtime/runtime-mode.ts | 7 + .../status-decision-committer.ts | 174 +++++-- 13 files changed, 1272 insertions(+), 168 deletions(-) create mode 100644 server/src/services/native-runtime/native-workspace-finalizer.ts diff --git a/packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md b/packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md index 1847ad651d8..4d73fc7f4db 100644 --- a/packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md +++ b/packages/paperclip-runner/docs/design/phase-6-thin-paperclip-adapter.md @@ -823,6 +823,46 @@ changing its facts or decision, suppressing its target mutation, duplicating a delivery, or restoring a synthetic `issue_checkout` fallback fails the focused database gate. +### Remediation 5 live-entrypoint proof amendment (2026-08-09) + +The remediation stays within the approved server authority seams and adds one +owning workspace service: + +```text +server/src/services/heartbeat.ts +server/src/services/native-runtime/native-session-executor.ts +server/src/services/native-runtime/native-interaction-bridge.ts +server/src/services/native-runtime/native-finalization-reconciler.ts +server/src/services/native-runtime/native-workspace-finalizer.ts +server/src/services/native-runtime/status-decision-committer.ts +server/src/services/native-runtime/runtime-mode.ts +server/src/__tests__/native-status-arbiter-corpus.test.ts +``` + +Policy resolvers remain pure and may be tested directly, but their return value +is not operational proof. Cancellation fixtures enter through +`cancelNativeSession`, which commits the decision or an explicit audit-only +replacement-turn outcome. Attention fixtures enter through +`routeNativeAttention`, which resolves a same-company eligible delegate and +uses the company-scoped issue service. REC-04/06/07/08 fixtures enter through +`reconcileNativeFinalizations`; REC-04 invokes the workspace operation recorder +and observes the actual finalizer result. REC-06/07/08 reclassify the persisted +result and contract into a new append-only assessment before an explicitly +authorized superseding commit; REC-07 cannot be selected by an unrelated work +product. MIG-08 persists the kill switch in the agent-owned runtime profile, +and `resolveNativeRuntimeMode` enforces that state for subsequent dispatches. + +`resolveNativeFinalizerStatus`, `resolveNativeAttentionStatus`, +`resolveNativeCancellationStatus`, `resolveNativeReconciliationStatus`, the +compatibility/migration resolvers, and their read models are policy evidence +only when called directly. A fixture claiming an operational effect must also +contain the named live-entrypoint receipt and the owning service's durable +target. Pending-effect replay remains acknowledgement-only and is restricted +to the original company, issue, decision, and still-existing target. Negative +tests deliberately remove each live entrypoint/action and the replay target; +the mapped fixture fails even though the corresponding pure resolver still +returns its expected label. + ## Commands the implementation must make runnable These commands are the acceptance contract for the implementation issue. They diff --git a/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md b/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md index 8d31da4362e..39eceb73968 100644 --- a/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md +++ b/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md @@ -112,11 +112,23 @@ persisted target, replay is required to retain the same decision with one delivery attempt, and an unknown-effect test proves the transaction leaves no decision, ledger, issue-version, or coordinator mutation. -The production call graph invokes the fact consumers from heartbeat -finalization, interaction-response projection, normalized session -cancellation, finalization reconciliation, and runtime compatibility/migration -selection. The corpus uses those same exported entry points and fails its row -when their returned semantics or materialized target state changes. +Operational rows use the production call graph, not resolver calls as a proxy. +Cancellation is committed and audited by the normalized session cancellation +entrypoint. Attention resolves an eligible same-company target and delegates +through the issue service. Reconciliation selects the persisted +status/evidence/policy branches and executes workspace recovery through the +workspace operation recorder. The rollout kill switch is stored in the owning +agent profile and consumed by subsequent runtime selection. Pending replay +only acknowledges its original decision-scoped target after verifying that the +target still exists. + +Direct calls to the finalizer, attention, cancellation, reconciliation, +compatibility, and migration resolvers are explicitly pure policy/read-model +tests. They do not satisfy an operational fixture without a live-entrypoint +receipt and concrete target state. Sabotage cases remove cancellation, +attention, reconciliation, rollout enforcement, and a pending replay target; +each mapped fixture fails while its pure resolver continues to return the +expected label. Result-less same-run recovery is also proven through the production heartbeat seam: a persisted envelope/checkpoint is leased by `reapOrphanedRuns`, enters diff --git a/packages/paperclip-runner/docs/tutorials/phase-06-thin-paperclip-adapter.md b/packages/paperclip-runner/docs/tutorials/phase-06-thin-paperclip-adapter.md index 41be67ec18b..91dde775220 100644 --- a/packages/paperclip-runner/docs/tutorials/phase-06-thin-paperclip-adapter.md +++ b/packages/paperclip-runner/docs/tutorials/phase-06-thin-paperclip-adapter.md @@ -60,7 +60,7 @@ pnpm --filter @paperclipai/server exec vitest run \ ``` This runs the unchanged package conformance suite against the real Paperclip -port. Forty-two targeted tests check company binding, duplicate/gap replay, +port. Forty-three targeted tests check company binding, duplicate/gap replay, checkpoint recovery, immutable result ingestion, durable-evidence authority, workspace-gated finalization, atomic liveness rollback, status-version CAS, migration repair, legacy byte equivalence, typed interaction materialization, @@ -68,19 +68,20 @@ same-run pre-result resumption, lease races, and audit output. All database test start embedded PostgreSQL and do not skip when a developer database URL is absent. -The corpus cases dispatch fixture-specific database shapes to named production -consumers and derive all eleven expected fields from their returns and -persisted rows: run status, status/preserve action, reason, required and -forbidden effects, live-path kind, claim preservation, native-record behavior, -decision count, maximum wake count, and maximum notification count. Every one -of the 70 matrix rows requires evidence from its responsible finalizer, -terminal projection, attention, cancellation, committer, reconciliation, -compatibility, or migration consumer; a missing call or changed return fails -the row. Each native effect must also have a delivered, company-bound target -row or concrete target-state mutation, and replay must retain one decision and -one delivery attempt. A separate unknown-effect case proves the status -transaction fails closed without partial rows. The recovery case uses a -scripted provider only at the backend +The corpus cases dispatch fixture-specific database shapes through live +cancellation, attention, reconciliation, and rollout entrypoints whenever an +operational effect is claimed. Workspace recovery records the actual operation +result, delegation uses an eligible same-company agent and the issue service, +and a persisted kill switch changes the next runtime selection. Pure resolver +and read-model calls remain useful policy tests but cannot satisfy these rows. +Each native effect must have a delivered, company-bound target or concrete +target-state mutation, and pending replay may only acknowledge its original +decision-scoped target. Sabotage tests remove each live action and the replay +target and require failure even while the pure resolver label remains correct. + +All eleven expected fields and all 70 matrix-row joins are still checked, and +an unknown-effect case proves the status transaction fails closed without +partial rows. The recovery case uses a scripted provider only at the backend boundary while the real heartbeat reaper, lease claim, original run execution, persistence port, finalizer, and terminal projection execute. diff --git a/packages/paperclip-runner/knowledge/evidence/2026-08-09-phase-06-verification.md b/packages/paperclip-runner/knowledge/evidence/2026-08-09-phase-06-verification.md index 166104d7f62..952501a145d 100644 --- a/packages/paperclip-runner/knowledge/evidence/2026-08-09-phase-06-verification.md +++ b/packages/paperclip-runner/knowledge/evidence/2026-08-09-phase-06-verification.md @@ -13,7 +13,7 @@ This record verifies the default-off Phase 6 tracer at the approved `ControlPlanePort`/`NativeSessionBackend` boundary. No browser surface changed, so Phase 6 has command/database evidence rather than screenshots. -The counts below are the PAP-16878 remediation rerun. “Internal canary” means +The counts below are the PAP-16879 remediation rerun. “Internal canary” means the selected-task database test through the public package session contract; it does not mean that a new live Codex provider task was dispatched. @@ -39,7 +39,7 @@ existing accepted result event, and appended only the missing terminal fact. # Real Paperclip adapter and public-session task -The complete acceptance gate below passed 42 tests in thirteen files against +The complete acceptance gate below passed 43 tests in thirteen files against embedded PostgreSQL with zero skipped tests. Database-backed tests start their own database instead of skipping when a developer `DATABASE_URL` is absent. The same package conformance suite passed unchanged against @@ -82,7 +82,7 @@ The focused corpus also proved: bindings, recovery, status, decisions, and effects roll back together before retry ownership is recorded. -The acceptance-matrix entry points passed as one 13-file, 42-test gate: +The acceptance-matrix entry points passed as one 13-file, 43-test gate: ```sh pnpm --filter @paperclipai/server exec vitest run \ @@ -103,20 +103,24 @@ pnpm --filter @paperclipai/server exec vitest run \ This gate proves concurrent event allocation, duplicate-only migration repair, bounded retry exhaustion, actual flag-off legacy execution, and executable -Section 18.13 fixture-to-consumer traceability. Each of the 52 fixtures creates -its own database shape and executes named production consumers. Their return -values and persisted rows supply all eleven expected fields: run status, -status/preserve action, reason, required and forbidden effects, live-path kind, -claim preservation, native-record behavior, decision count, maximum wake -count, and maximum notification count. Each of the 70 matrix rows has an -explicit responsible finalizer, terminal projection, attention, cancellation, -committer, reconciliation, compatibility, or migration consumer; the row fails -if that consumer was not executed or its returned semantics differ. A mutation -test changes every expected field for every fixture and proves comparison -failure. Every required native effect is joined to its actual target state, -each decision replay retains one identity and one delivery attempt, and an -unknown effect rolls the entire transaction back. No test-owned scenario -policy supplies an observation. The selected-task +Section 18.13 fixture-to-consumer traceability. All 52 fixtures and 70 rows are +classified as either live operational proofs or explicit pure policy/read-model +checks. Cancellation persists its decision or audit-only receipt through +`cancelNativeSession`; attention resolves a same-company eligible target and +uses the issue service; REC-04/06/07/08 enter the reconciler, with REC-04 +recording an observed workspace operation and REC-06/07/08 writing a newly +classified append-only assessment before superseding the prior decision; and +MIG-08 is enforced by the next runtime selection from the agent-owned +kill-switch profile. + +All eleven expected fields remain derived from entrypoint returns and durable +rows. Every required native effect is joined to its owning target state, each +decision replay retains one identity and one delivery attempt, and pending +replay verifies the original company/issue/decision target before acknowledging +it. The negative suite removes each live action and deletes a replay target; +the mapped fixtures fail while the direct policy resolver still returns the +expected label. An unknown effect rolls the entire transaction back. No +test-owned scenario policy supplies an observation. The selected-task and recovery canaries remain scripted only at the provider boundary, while the production Paperclip persistence/finalization paths and the flag-off legacy adapter/finalizer path execute for real. diff --git a/packages/paperclip-runner/knowledge/journal/2026-08-09-phase-06-implementation.md b/packages/paperclip-runner/knowledge/journal/2026-08-09-phase-06-implementation.md index 500b9c7fe7d..1645dec5e41 100644 --- a/packages/paperclip-runner/knowledge/journal/2026-08-09-phase-06-implementation.md +++ b/packages/paperclip-runner/knowledge/journal/2026-08-09-phase-06-implementation.md @@ -57,6 +57,16 @@ existing server authority and keep legacy execution as the default. 15. Prove at-most-once materialization by replaying each corpus decision and requiring one decision identity, one delivery attempt, and the persisted target state. +16. Treat direct resolver calls as pure policy/read-model evidence only. + Operational corpus rows must enter through cancellation, attention, + reconciliation, or runtime selection and retain that entrypoint receipt. +17. Resume workspace finalization through the workspace operation recorder and + observe its actual result; never insert a pre-succeeded recovery marker. +18. Resolve attention delegates from eligible same-company agents and create + delegated work through the issue service. Persist authorized status and + rollout changes through their owning issue and agent services. +19. Keep pending replay acknowledgement-only, scoped to its original decision, + and fail when the owning transaction's target no longer exists. # Evidence @@ -64,7 +74,7 @@ existing server authority and keep legacy execution as the default. - [Phase 6 reference](../../docs/phase-06-thin-paperclip-adapter.md) - [Runnable tutorial](../../docs/tutorials/phase-06-thin-paperclip-adapter.md) - Package conformance and recovery: four files, seven tests passed. -- Phase 6 acceptance-matrix entry points: thirteen files, forty-two tests passed +- Phase 6 acceptance-matrix entry points: thirteen files, forty-three tests passed with zero skips, including the database-backed selected-task canary, six atomic-liveness failpoints, migration, sequencing, bounded recovery, and legacy compatibility. @@ -83,6 +93,10 @@ existing server authority and keep legacy execution as the default. recovery, run, workspace-operation, contract, decision, or governance state. Unknown effects fail before any decision, ledger, status-version, or coordinator mutation. +- Negative entrypoint proofs remove cancellation auditing, attention routing, + reconciliation/workspace execution, rollout selector enforcement, and the + original pending-effect target. Every mapped fixture fails even though its + pure resolver still returns the expected policy label. - The production heartbeat reaper resumed one persisted result-less run through the original execution/finalization path with the flag disabled. It recovered an already-active scripted provider turn and produced exactly one result, diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 50dd739b4d1..85e493aef4a 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -27,6 +27,7 @@ import { applyNativeAttentionStatusDecision, materializeNativeInteractionResponses, rejectUnsupportedNativeRuntimeRequest, + routeNativeAttention, resolveNativeAttentionStatus, } from "../services/native-runtime/native-interaction-bridge.js"; import { @@ -109,6 +110,8 @@ type FixtureObservation = PolicyObservation & { consumerEvidenceByRow: Map; }; +type DisabledLiveEntrypoint = "attention" | "cancellation" | "reconciliation" | "rollout"; + const corpusPath = fileURLToPath(new URL( "../../../packages/paperclip-runner/spec/fixtures/status-authority-phase5.json", import.meta.url, @@ -239,6 +242,10 @@ const supersedingDecisionStates = new Set([ "board_cancelled_before_cas", "new_policy_requires_review", "authorized_writer_incremented_version", ]); +const liveReconciliationStates = new Set([ + "board_cancelled_before_cas", "new_evidence_satisfies_contract", "new_policy_requires_review", +]); + function initialRunStatus(fixture: Fixture) { const terminalState = fixture.given.runTerminalState; if (["succeeded", "failed", "cancelled", "active"].includes(String(terminalState))) { @@ -362,18 +369,39 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { let db: ReturnType; const companyId = randomUUID(); const agentId = randomUUID(); + const delegateAgentId = randomUUID(); + const outsideCompanyId = randomUUID(); + const outsideAgentId = randomUUID(); beforeAll(async () => { temporary = await startEmbeddedPostgresTestDatabase("paperclip-status-corpus-"); db = createDb(temporary.connectionString); await db.insert(companies).values({ id: companyId, name: "Status corpus", issuePrefix: "PSC" }); - await db.insert(agents).values({ - id: agentId, - companyId, - name: "Status corpus agent", - adapterType: "codex_local", - status: "running", - }); + await db.insert(companies).values({ id: outsideCompanyId, name: "Outside corpus", issuePrefix: "OUT" }); + await db.insert(agents).values([ + { + id: agentId, + companyId, + name: "Status corpus agent", + adapterType: "codex_local", + status: "running", + }, + { + id: delegateAgentId, + companyId, + name: "Eligible status corpus delegate", + adapterType: "codex_local", + status: "idle", + capabilities: "Provide production status-authority expertise", + }, + { + id: outsideAgentId, + companyId: outsideCompanyId, + name: "Outside-company agent", + adapterType: "codex_local", + status: "idle", + }, + ]); }, 30_000); afterAll(async () => temporary?.cleanup()); @@ -385,6 +413,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { const resultId = randomUUID(); const assessmentId = randomUUID(); const workProductId = randomUUID(); + const assessmentCreatedAt = new Date(); const priorStatus = String(fixture.given.priorIssueStatus ?? "in_progress"); const completionState = String(fixture.given.completionState ?? ""); const nativeRecords = fixture.mode === "native" && !noNativeRecordStates.has(completionState); @@ -415,6 +444,8 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { runtimeMode: "native", runtimeModeResolvedAt: new Date(), contextSnapshot: { issueId, fixtureId: fixture.id }, + completionContractId: contractId, + completionContractSha256: `contract:${fixture.id}`, }); await db.insert(completionContracts).values({ id: contractId, @@ -441,6 +472,30 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { schemaStatus: "accepted", resultJson: { fixtureId: fixture.id, + result: { + reportedWorkDisposition: fixtureDisposition(fixture), + summary: fixture.id, + completionClaim: { + contractRevision: "corpus-v1", + objectiveSatisfied: true, + criteria: [{ + criterionId: "objective", + status: "satisfied", + evidenceRefs: [`work_product:${workProductId}`], + }], + remainingWork: [], + }, + verification: [{ + commandOrCheck: "fixture", + status: "passed", + artifactRef: `work_product:${workProductId}`, + }], + }, + terminal: { + runTerminalState: fixture.given.runTerminalState === "failed" + ? "failed" + : fixture.given.runTerminalState === "cancelled" ? "cancelled" : "succeeded", + }, ...(fixture.given.reportedWorkDisposition === null && fixture.given.nativeFinalization !== "invalid" ? {} : { completionClaim: { fixtureId: fixture.id, preserved: true } }), @@ -455,7 +510,9 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { provider: "paperclip", title: `${fixture.id} evidence`, status: "ready_for_review", - reviewState: "approved", + reviewState: completionState === "new_evidence_satisfies_contract" ? "none" : "approved", + createdAt: new Date(assessmentCreatedAt.getTime() - 1_000), + updatedAt: new Date(assessmentCreatedAt.getTime() - 1_000), }); await db.insert(workAssessments).values({ id: assessmentId, @@ -466,11 +523,15 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { resultId, triggerKind: "native_result", triggerActorCompanyId: companyId, - priorIssueStatus: priorStatus, + priorIssueStatus: completionState === "board_cancelled_before_cas" ? "in_progress" : priorStatus, priorStatusVersion: 0, - policyVersion: "phase6-v2", - assessmentJson: { fixtureId: fixture.id }, + policyVersion: completionState === "new_policy_requires_review" ? "phase6-v1" : "phase6-v2", + assessmentJson: { + fixtureId: fixture.id, + ...(completionState === "new_evidence_satisfies_contract" ? { allCriteriaSatisfied: false } : {}), + }, inputDigest: `assessment:${fixture.id}`, + createdAt: assessmentCreatedAt, }); await db.insert(nativeRunFinalizations).values({ runId, @@ -483,7 +544,10 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { return { issueId, runId, workProductId, nativeRecords, assessmentId, contractId, resultId }; } - async function executeFixture(fixture: Fixture): Promise { + async function executeFixture( + fixture: Fixture, + options: { disableLiveEntrypoint?: DisabledLiveEntrypoint } = {}, + ): Promise { const completionState = String(fixture.given.completionState ?? ""); const seeded = await seedFixture(fixture); const consumerExecutions: ConsumerExecution[] = []; @@ -536,22 +600,134 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { let semanticConsumer: string | null = null; let consumerDecision: NativeStatusDecision | null = null; + let liveEntrypointCommitted = false; + let liveAttentionInteractionId: string | null = null; + const attentionFacts = attentionFactsFor(completionState, fixture.id, governanceGate); + if ( + seeded.nativeRecords + && options.disableLiveEntrypoint !== "attention" + && attentionFacts + && ["attention_response", "attention_candidate", "interaction", "monitor"].includes(String(fixture.given.trigger)) + ) { + let canonicalRequestId: string | undefined; + if (attentionFacts.route === "duplicate") { + canonicalRequestId = randomUUID(); + liveAttentionInteractionId = randomUUID(); + await db.insert(issueThreadInteractions).values([ + { + id: canonicalRequestId, + companyId, + issueId: seeded.issueId, + kind: "request_confirmation", + status: "pending", + idempotencyKey: `canonical:${fixture.id}`, + payload: { version: 1, prompt: `Canonical ${fixture.id}` }, + }, + { + id: liveAttentionInteractionId, + companyId, + issueId: seeded.issueId, + kind: "request_confirmation", + status: "expired", + resolvedAt: new Date(), + idempotencyKey: `duplicate:${fixture.id}`, + payload: { version: 1, prompt: `Duplicate ${fixture.id}` }, + result: { version: 1, outcome: "superseded_by_newer_request", supersededByInteractionId: canonicalRequestId }, + }, + ]); + } else if (attentionFacts.responseState === "stale") { + liveAttentionInteractionId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: liveAttentionInteractionId, + companyId, + issueId: seeded.issueId, + kind: "request_confirmation", + status: "expired", + resolvedAt: new Date(), + payload: { version: 1, prompt: `Stale ${fixture.id}` }, + result: { version: 1, outcome: "superseded_by_comment", commentId: randomUUID() }, + }); + } else if (attentionFacts.responseState === "expired") { + liveAttentionInteractionId = randomUUID(); + await db.insert(issueThreadInteractions).values({ + id: liveAttentionInteractionId, + companyId, + issueId: seeded.issueId, + kind: "ask_user_questions", + status: "pending", + payload: { version: 1, questions: [] }, + }); + } + const routed = await routeNativeAttention({ + db, + runId: seeded.runId, + candidate: { + route: attentionFacts.route, + summary: attentionFacts.summary, + responseState: attentionFacts.responseState, + budgetExhausted: attentionFacts.budgetExhausted, + governanceGate: attentionFacts.governanceGate, + targetAgentId: completionState === "cross_company_target" ? outsideAgentId : null, + targetInteractionId: liveAttentionInteractionId ?? undefined, + canonicalRequestId, + }, + }); + semanticConsumer = "native-attention-resolver"; + consumerDecision = pushDecisionConsumer(semanticConsumer, routed.decision); + liveEntrypointCommitted = true; + for (const target of routed.targets) materializedEffects.add(target.effectKind); + consumerExecutions.push({ + consumer: "native-attention-entrypoint", + observed: { + decisionId: routed.decisionId, + resolvedTargetAgentId: routed.resolvedTargetAgentId, + targetCount: routed.targets.length, + }, + }); + if (attentionFacts.route === "agent" && completionState !== "cross_company_target") { + expect(routed.resolvedTargetAgentId, `${fixture.id}:eligible delegate`).toBe(delegateAgentId); + } + } if (["turn_scope", "run_scope", "issue_scope_authorized", "replacement_turn_accepted"].includes(completionState)) { semanticConsumer = "native-cancellation-authority"; const scope = completionState === "run_scope" ? "run" : completionState === "issue_scope_authorized" ? "issue" : "turn"; - consumerDecision = pushDecisionConsumer(semanticConsumer, resolveNativeCancellationStatus({ - scope, - priorIssueStatus, - agentId, - replacementAccepted: completionState === "replacement_turn_accepted", - })); + if (options.disableLiveEntrypoint === "cancellation") { + consumerDecision = pushDecisionConsumer(semanticConsumer, resolveNativeCancellationStatus({ + scope, + priorIssueStatus, + agentId, + replacementAccepted: completionState === "replacement_turn_accepted", + })); + } else { + const cancellation = await cancelNativeSession(seeded.runId, `fixture:${fixture.id}`, { + db, + scope, + replacementAccepted: completionState === "replacement_turn_accepted", + }); + if (typeof cancellation === "boolean" || !cancellation.decision || !cancellation.auditId) { + throw new Error(`${fixture.id}: live cancellation did not persist an authoritative outcome`); + } + consumerDecision = pushDecisionConsumer(semanticConsumer, cancellation.decision); + liveEntrypointCommitted = true; + for (const effect of cancellation.decision.effects) materializedEffects.add(effect.kind); + consumerExecutions.push({ + consumer: "native-session-cancellation", + observed: { + dispatched: cancellation.dispatched, + scope, + reasonCode: cancellation.decision.reasonCode, + decisionId: cancellation.decisionId, + auditId: cancellation.auditId, + }, + }); + } } else { const reconciliationFacts = reconciliationFactsFor(completionState); - const attentionFacts = attentionFactsFor(completionState, fixture.id, governanceGate); const compatibilityFacts = compatibilityFactsFor(completionState); const migrationFacts = migrationFactsFor(completionState); if ( reconciliationFacts + && !liveReconciliationStates.has(completionState) && ( supersedingDecisionStates.has(completionState) || (fixture.covers.reconciliationRows ?? []).length > 0 @@ -564,10 +740,12 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { priorIssueStatus, agentId, })); - } else if (attentionFacts && ["attention_response", "attention_candidate", "interaction", "monitor"].includes(String(fixture.given.trigger))) { + } else if (!consumerDecision && attentionFacts && ["attention_response", "attention_candidate", "interaction", "monitor"].includes(String(fixture.given.trigger))) { semanticConsumer = "native-attention-resolver"; consumerDecision = pushDecisionConsumer(semanticConsumer, resolveNativeAttentionStatus({ - facts: attentionFacts, + facts: options.disableLiveEntrypoint === "attention" && attentionFacts.route === "agent" + ? { ...attentionFacts, resolvedTargetAgentId: delegateAgentId } + : attentionFacts, priorIssueStatus, agentId, })); @@ -598,7 +776,12 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { if (attentionFacts && (fixture.covers.attentionRows ?? []).length > 0 && semanticConsumer !== "native-attention-resolver") { pushDecisionConsumer("native-attention-resolver", resolveNativeAttentionStatus({ facts: attentionFacts, priorIssueStatus, agentId })); } - if (reconciliationFacts && (fixture.covers.reconciliationRows ?? []).length > 0 && semanticConsumer !== "native-reconciliation-consumer") { + if ( + reconciliationFacts + && !liveReconciliationStates.has(completionState) + && (fixture.covers.reconciliationRows ?? []).length > 0 + && semanticConsumer !== "native-reconciliation-consumer" + ) { pushDecisionConsumer("native-reconciliation-consumer", resolveNativeReconciliationStatus({ facts: reconciliationFacts, priorIssueStatus, agentId })); } if (compatibilityFacts && (fixture.covers.compatibilityRows ?? []).some((row) => ["COMP-02", "COMP-05", "COMP-06", "COMP-07", "COMP-08"].includes(row)) && semanticConsumer !== "native-compatibility-status") { @@ -609,6 +792,69 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { } } + if ( + seeded.nativeRecords + && options.disableLiveEntrypoint !== "reconciliation" + && liveReconciliationStates.has(completionState) + ) { + const [priorDecision] = await db.insert(statusDecisions).values({ + companyId, + issueId: seeded.issueId, + assessmentId: seeded.assessmentId, + decisionVersion: 1, + policyVersion: completionState === "new_policy_requires_review" ? "phase6-v1" : "phase6-v2", + fromStatus: completionState === "board_cancelled_before_cas" ? "in_progress" : priorIssueStatus, + toStatus: completionState === "board_cancelled_before_cas" ? "in_progress" : priorIssueStatus, + reasonCode: "prior_fixture_decision", + decisionJson: { superseded: true }, + decisionDigest: `prior-decision:${fixture.id}`, + applicationState: "applied", + appliedAt: new Date(), + }).returning({ id: statusDecisions.id }); + await db.update(issues).set({ + statusVersion: 1, + lastStatusDecisionId: priorDecision!.id, + }).where(eq(issues.id, seeded.issueId)); + await db.update(nativeRunFinalizations).set({ + phase: "committed", + decisionId: priorDecision!.id, + }).where(eq(nativeRunFinalizations.runId, seeded.runId)); + if (completionState === "new_evidence_satisfies_contract") { + await db.update(issueWorkProducts).set({ + reviewState: "approved", + updatedAt: new Date(Date.now() + 1_000), + }).where(eq(issueWorkProducts.id, seeded.workProductId)); + } + const [reconciled] = await reconcileNativeFinalizations(db, [seeded.runId]); + if (!reconciled?.reconciliationDecision || !reconciled.decisionId) { + throw new Error(`${fixture.id}: live reconciliation did not commit an authoritative decision`); + } + semanticConsumer = "native-reconciliation-consumer"; + consumerDecision = pushDecisionConsumer(semanticConsumer, reconciled.reconciliationDecision); + liveEntrypointCommitted = true; + consumerExecutions.push({ + consumer: "native-reconciliation-entrypoint", + observed: { + action: reconciled.reconciliationAction, + decisionId: reconciled.decisionId, + }, + }); + } + if ( + seeded.nativeRecords + && options.disableLiveEntrypoint === "reconciliation" + && liveReconciliationStates.has(completionState) + ) { + const facts = reconciliationFactsFor(completionState); + if (!facts) throw new Error(`${fixture.id}: missing reconciliation facts`); + semanticConsumer = "native-reconciliation-consumer"; + consumerDecision = pushDecisionConsumer(semanticConsumer, resolveNativeReconciliationStatus({ + facts, + priorIssueStatus, + agentId, + })); + } + let assessment: NativeEvidenceAssessment | null = null; if (seeded.nativeRecords && ["runner_finalizer", "dependency", "shadow_comparator", "read_model", "authorized_agent"].includes(String(fixture.given.trigger))) { const accepted = completeEvidenceStates.has(completionState); @@ -679,7 +925,11 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { } } - if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "resume_workspace_operation")) { + if ( + seeded.nativeRecords + && options.disableLiveEntrypoint !== "reconciliation" + && consumerDecision?.effects.some((effect) => effect.kind === "resume_workspace_operation") + ) { await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: seeded.runId, @@ -687,11 +937,25 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { phase: "workspace_finalize", status: "failed", exitCode: 1, + cwd: process.cwd(), finishedAt: new Date(), }); + const [reconciled] = await reconcileNativeFinalizations(db, [seeded.runId]); + if (!reconciled || reconciled.reconciliationAction !== "resume_workspace_operation") { + throw new Error(`${fixture.id}: live workspace reconciliation did not execute (${JSON.stringify(reconciled)})`); + } + liveEntrypointCommitted = true; + materializedEffects.add("resume_workspace_operation"); + consumerExecutions.push({ + consumer: "native-reconciliation-entrypoint", + observed: { + action: reconciled.reconciliationAction, + workspaceOperationId: reconciled.workspaceOperationId, + workspaceFinalizeStatus: reconciled.workspaceFinalizeStatus, + }, + }); } - let liveAttentionInteractionId: string | null = null; - if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "link_canonical_request")) { + if (!liveEntrypointCommitted && seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "link_canonical_request")) { const canonicalInteractionId = randomUUID(); liveAttentionInteractionId = randomUUID(); await db.insert(issueThreadInteractions).values([ @@ -721,7 +985,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }, ]); } - if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "record_stale_response")) { + if (!liveEntrypointCommitted && seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "record_stale_response")) { liveAttentionInteractionId = randomUUID(); await db.insert(issueThreadInteractions).values({ id: liveAttentionInteractionId, @@ -734,7 +998,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { result: { version: 1, outcome: "superseded_by_comment", commentId: randomUUID() }, }); } - if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "record_expiry")) { + if (!liveEntrypointCommitted && seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "record_expiry")) { await db.insert(issueThreadInteractions).values({ companyId, issueId: seeded.issueId, @@ -797,6 +1061,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { if ( seeded.nativeRecords && zeroDecisionStates.has(completionState) + && !liveEntrypointCommitted && consumerDecision && consumerDecision.effects.some((effect) => [ "link_canonical_request", @@ -842,6 +1107,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { && consumerDecision.reasonCode !== null && completionState !== "replacement_turn_accepted" && !zeroDecisionStates.has(completionState) + && !liveEntrypointCommitted ) { let priorStatusVersion = 0; let priorDecisionId: string | null = null; @@ -925,6 +1191,12 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { } } + if (options.disableLiveEntrypoint === "rollout" && completionState === "kill_switch_during_active_native_run") { + await db.update(agents).set({ + runtimeConfig: { nativeRunner: { mode: "native", backend: "codex_app_server", protocolVersion: 1 } }, + }).where(eq(agents.id, agentId)); + } + if (seeded.nativeRecords && (rolledBack || completionState === "safe_partial_parse")) { const failure = await recordNativeFinalizationFailure({ db, @@ -978,28 +1250,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }); } - if (["turn_scope", "run_scope", "issue_scope_authorized", "replacement_turn_accepted"].includes(completionState)) { - const cancellation = await cancelNativeSession(seeded.runId, `fixture:${fixture.id}`, { - db, - scope: completionState === "run_scope" ? "run" : completionState === "issue_scope_authorized" ? "issue" : "turn", - replacementAccepted: completionState === "replacement_turn_accepted", - }); - if (completionState === "replacement_turn_accepted" && cancellation.decision?.effects.some((effect) => effect.kind === "accept_replacement_turn")) { - materializedEffects.add("accept_replacement_turn"); - } - consumerExecutions.push({ - consumer: "native-session-cancellation", - observed: { - dispatched: cancellation.dispatched, - scope: completionState, - reasonCode: cancellation.decision?.reasonCode ?? null, - }, - }); - } - - const boardTarget = completionState === "issue_scope_authorized" - ? "cancelled" - : completionState === "explicit_resume_capability" + const boardTarget = completionState === "explicit_resume_capability" ? "in_progress" : completionState === "authorized_status_write" ? "in_review" @@ -1127,6 +1378,10 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { const target = await db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests) .where(eq(agentWakeupRequests.id, effectRow.targetId!)).then((rows) => rows[0] ?? null); expect(target, `${fixture.id}:${effectRow.effectKind}:wake-target`).not.toBeNull(); + } else if (effectRow.targetType === "agent") { + const target = await db.select({ id: agents.id }).from(agents) + .where(eq(agents.id, effectRow.targetId!)).then((rows) => rows[0] ?? null); + expect(target, `${fixture.id}:${effectRow.effectKind}:agent-target`).not.toBeNull(); } else if (effectRow.targetType === "delegated_issue" || effectRow.targetType === "issue" || effectRow.targetType === "issue_checkout" || effectRow.targetType === "issue_unblock_descriptor") { const target = await db.select({ id: issues.id }).from(issues) .where(eq(issues.id, effectRow.targetId!)).then((rows) => rows[0] ?? null); @@ -1182,7 +1437,19 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { if (persistedEffects.includes("record_mode_native")) expect(persistedRun?.runtimeMode, `${fixture.id}:mode target`).toBe("native"); if (persistedEffects.includes("record_policy_version")) expect(persistedRunnerProfile.nativeStatusPolicyVersion, `${fixture.id}:policy target`).toBe("phase6-v2"); if (persistedEffects.includes("finish_as_native")) expect(persistedRunResult.nativeKillSwitchDisposition, `${fixture.id}:finish-native target`).toBe("finish_as_native"); - if (persistedEffects.includes("stop_new_native_dispatch")) expect(persistedRunnerProfile.nativeDispatchAfterRun, `${fixture.id}:kill-switch target`).toBe("legacy"); + if (persistedEffects.includes("stop_new_native_dispatch")) { + const runtimeConfig = await db.select({ runtimeConfig: agents.runtimeConfig }).from(agents) + .where(eq(agents.id, agentId)).then((rows) => rows[0]?.runtimeConfig ?? {}); + expect((runtimeConfig as { nativeRunner?: { mode?: string } }).nativeRunner?.mode, `${fixture.id}:kill-switch target`).toBe("legacy"); + expect(resolveNativeRuntimeMode({ + enabled: true, + runtimeConfig, + agent: { id: agentId, status: "running", adapterType: "codex_local" }, + issue: { id: seeded.issueId, workMode: "standard" }, + target: { kind: "local" }, + workspaceId: "fixture-workspace", + })).toEqual(expect.objectContaining({ kind: "legacy", reason: "native_kill_switch_persisted" })); + } if (materializedEffects.has("link_canonical_request")) { expect(interactionRows.some((row) => row.summary?.startsWith("Canonical native attention request:")), `${fixture.id}:canonical link target`).toBe(true); } @@ -1319,6 +1586,27 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }, }); if (consumerExecutions.length < 2) throw new Error(`${fixture.id} did not execute a concern consumer`); + if ( + ["turn_scope", "run_scope", "issue_scope_authorized", "replacement_turn_accepted"].includes(completionState) + && !consumerExecutions.some((execution) => + execution.consumer === "native-session-cancellation" && typeof execution.observed.auditId === "string" + ) + ) { + throw new Error(`${fixture.id}: live cancellation proof missing`); + } + if ( + attentionFacts + && ["attention_response", "attention_candidate", "interaction", "monitor"].includes(String(fixture.given.trigger)) + && !consumerExecutions.some((execution) => execution.consumer === "native-attention-entrypoint") + ) { + throw new Error(`${fixture.id}: live attention proof missing`); + } + if ( + ["REC-04", "REC-06", "REC-07", "REC-08"].some((row) => (fixture.covers.reconciliationRows ?? []).includes(row)) + && !consumerExecutions.some((execution) => execution.consumer === "native-reconciliation-entrypoint") + ) { + throw new Error(`${fixture.id}: live reconciliation proof missing`); + } return { fixtureId: fixture.id, @@ -1418,6 +1706,109 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { } }, 60_000); + it("fails mapped fixtures when live entrypoints or owning actions are removed despite matching policy labels", async () => { + const byId = (id: string) => { + const fixture = corpus.fixtures.find((candidate) => candidate.id === id); + if (!fixture) throw new Error(`fixture missing: ${id}`); + return fixture; + }; + + expect(resolveNativeCancellationStatus({ + scope: "turn", + priorIssueStatus: "in_progress", + agentId, + }).reasonCode).toBe("cancellation_turn_only"); + await expect(executeFixture(byId("turn-only-cancellation"), { + disableLiveEntrypoint: "cancellation", + })).rejects.toThrow("live cancellation proof missing"); + + expect(resolveNativeAttentionStatus({ + facts: { + companyScopeValid: true, + responseState: "none", + route: "agent", + summary: "delegate", + resolvedTargetAgentId: delegateAgentId, + }, + priorIssueStatus: "in_progress", + agentId, + }).reasonCode).toBe("attention_routed_to_agent"); + await expect(executeFixture(byId("human-request-routed-to-agent"), { + disableLiveEntrypoint: "attention", + })).rejects.toThrow("live attention proof missing"); + + expect(resolveNativeReconciliationStatus({ + facts: { workspaceOperationPending: true }, + priorIssueStatus: "in_progress", + agentId, + }).effects.map((effect) => effect.kind)).toContain("resume_workspace_operation"); + await expect(executeFixture(byId("crash-before-workspace-finalization"), { + disableLiveEntrypoint: "reconciliation", + })).rejects.toThrow("native_workspace_operation_not_executed"); + + expect(resolveNativeMigrationStatus({ + facts: { killSwitchActiveForNewRuns: true }, + priorIssueStatus: "in_progress", + agentId, + }).effects.map((effect) => effect.kind)).toContain("stop_new_native_dispatch"); + await expect(executeFixture(byId("migration-kill-switch-rollback"), { + disableLiveEntrypoint: "rollout", + })).rejects.toThrow("kill-switch target"); + + const pendingFixture = byId("crash-after-decision-commit"); + const seeded = await seedFixture(pendingFixture); + const [decision] = await db.insert(statusDecisions).values({ + companyId, + issueId: seeded.issueId, + assessmentId: seeded.assessmentId, + decisionVersion: 1, + policyVersion: "phase6-v2", + fromStatus: "in_progress", + toStatus: "in_progress", + reasonCode: "live_continuation_registered", + decisionJson: { fixtureId: pendingFixture.id }, + decisionDigest: `pending-negative:${seeded.issueId}`, + applicationState: "applied", + appliedAt: new Date(), + }).returning({ id: statusDecisions.id }); + const [wake] = await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_status_changed", + payload: { issueId: seeded.issueId, nativeDecisionId: decision!.id }, + requestedByActorType: "system", + requestedByActorId: "native-status-committer", + idempotencyKey: `pending-negative:${seeded.issueId}`, + }).returning({ id: agentWakeupRequests.id }); + await db.insert(statusDecisionEffects).values({ + companyId, + issueId: seeded.issueId, + decisionId: decision!.id, + ordinal: 1, + effectKind: "enqueue_continuation", + targetType: "agent_wakeup_request", + targetId: wake!.id, + idempotencyKey: `pending-negative:${seeded.issueId}`, + payload: { fixtureId: pendingFixture.id }, + deliveryState: "pending", + attemptCount: 0, + }); + await db.update(nativeRunFinalizations).set({ + phase: "committed", + decisionId: decision!.id, + }).where(eq(nativeRunFinalizations.runId, seeded.runId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake!.id)); + expect(resolveNativeReconciliationStatus({ + facts: { undeliveredEffectCount: 1 }, + priorIssueStatus: "in_progress", + agentId, + }).effects.map((effect) => effect.kind)).toContain("dispatch_pending_effect"); + await expect(reconcileNativeFinalizations(db, [seeded.runId])) + .rejects.toThrow("native_pending_effect_target_missing:enqueue_continuation"); + }, 30_000); + it("fails the transaction closed for an unknown status effect", async () => { const fixture = corpus.fixtures.find((candidate) => candidate.mode === "native"); if (!fixture) throw new Error("native corpus fixture missing"); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index d8f8f63918b..18a8925b275 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -18737,7 +18737,10 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const running = runningProcesses.get(run.id); try { - await cancelNativeSession(run.id, reason, { db, scope: "run" }); + const nativeCancellation = await cancelNativeSession(run.id, reason, { db, scope: "run" }); + if (run.runtimeMode === "native" && (!nativeCancellation.decision || !nativeCancellation.auditId)) { + throw new Error("native_cancellation_outcome_not_audited"); + } if (running) { await terminateHeartbeatRunProcess({ pid: running.child.pid ?? run.processPid, @@ -18755,11 +18758,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) } const finishedAt = new Date(); + const persistedCancellationResult = run.runtimeMode === "native" + ? await getRun(run.id).then((current) => parseObject(current?.resultJson)) + : {}; const cancelled = await setRunStatus(run.id, "cancelled", { finishedAt, error: reason, errorCode, - ...(resultJson ? { resultJson } : {}), + ...(resultJson || Object.keys(persistedCancellationResult).length > 0 + ? { resultJson: { ...persistedCancellationResult, ...(resultJson ?? {}) } } + : {}), }); await setWakeupStatus(run.wakeupRequestId, "cancelled", { diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index d73dc9c36ac..e32d2492b69 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -2,15 +2,26 @@ import { createHash, randomUUID } from "node:crypto"; import { and, desc, eq, gt, inArray, isNotNull, isNull, lte, notInArray, or } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { + completionContracts, heartbeatRuns, + issueWorkProducts, issues, nativeRunFinalizations, + nativeRunResults, statusDecisionEffects, + statusDecisions, + workAssessments, workspaceOperations, } from "@paperclipai/db"; import { finalizeNativeRun, recordNativeFinalizationFailure } from "./native-run-finalizer.js"; -import { dispatchPendingNativeStatusEffects } from "./status-decision-committer.js"; +import { + commitNativeStatusDecision, + dispatchPendingNativeStatusEffects, +} from "./status-decision-committer.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; +import { resumeNativeWorkspaceFinalization } from "./native-workspace-finalizer.js"; +import { classifyNativeEvidence } from "./evidence-classifier.js"; +import { recordNativeWorkAssessment } from "./work-assessments.js"; import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, @@ -31,6 +42,12 @@ export type NativeReconciliationFacts = { statusVersionAdvanced?: boolean; }; +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? value as Record + : {}; +} + /** Append-only recovery/reconciliation re-arbitration from durable facts. */ export function resolveNativeReconciliationStatus(input: { facts: NativeReconciliationFacts; @@ -288,6 +305,9 @@ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) { agentId: heartbeatRuns.agentId, issueId: nativeRunFinalizations.issueId, issueStatus: issues.status, + issueStatusVersion: issues.statusVersion, + issueDecisionId: issues.lastStatusDecisionId, + assessmentId: nativeRunFinalizations.assessmentId, decisionId: nativeRunFinalizations.decisionId, }) .from(heartbeatRuns) @@ -340,6 +360,145 @@ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) { results.push({ runId: row.runId, phase: "committed", deliveredEffectIds }); continue; } + if (row.assessmentId) { + const assessment = await db.select({ + id: workAssessments.id, + contractId: workAssessments.contractId, + resultId: workAssessments.resultId, + policyVersion: workAssessments.policyVersion, + priorIssueStatus: workAssessments.priorIssueStatus, + priorStatusVersion: workAssessments.priorStatusVersion, + assessmentJson: workAssessments.assessmentJson, + createdAt: workAssessments.createdAt, + }).from(workAssessments).where(and( + eq(workAssessments.id, row.assessmentId), + eq(workAssessments.companyId, row.companyId), + eq(workAssessments.issueId, row.issueId), + )).limit(1).then((entries) => entries[0] ?? null); + const changedEvidence = assessment + ? await db.select({ id: issueWorkProducts.id }).from(issueWorkProducts).where(and( + eq(issueWorkProducts.companyId, row.companyId), + eq(issueWorkProducts.issueId, row.issueId), + or( + gt(issueWorkProducts.createdAt, assessment.createdAt), + gt(issueWorkProducts.updatedAt, assessment.createdAt), + ), + )).limit(1).then((entries) => entries[0] ?? null) + : null; + const policyVersionChanged = assessment?.policyVersion !== NATIVE_STATUS_ARBITER_POLICY_VERSION; + const currentDecision = row.decisionId + ? await db.select({ + assessmentId: statusDecisions.assessmentId, + decisionVersion: statusDecisions.decisionVersion, + toStatus: statusDecisions.toStatus, + decisionJson: statusDecisions.decisionJson, + }).from(statusDecisions).where(and( + eq(statusDecisions.id, row.decisionId), + eq(statusDecisions.companyId, row.companyId), + eq(statusDecisions.issueId, row.issueId), + )).limit(1).then((entries) => entries[0] ?? null) + : null; + const currentDecisionJson = record(currentDecision?.decisionJson); + const expectedProjectedVersion = currentDecision + ? Number(currentDecision.decisionVersion) - (currentDecisionJson.statusAction === "preserve" ? 1 : 0) + : null; + const issueMatchesCurrentDecision = !!assessment + && !!currentDecision + && row.issueDecisionId === row.decisionId + && currentDecision.assessmentId === assessment.id + && currentDecision.toStatus === row.issueStatus + && expectedProjectedVersion === Number(row.issueStatusVersion); + const authoritativeStatusChanged = !!assessment && !issueMatchesCurrentDecision && ( + assessment.priorIssueStatus !== row.issueStatus + || Number(assessment.priorStatusVersion) !== Number(row.issueStatusVersion) + ); + let reassessment = null; + let resultRow = null; + let contractRow = null; + if (assessment && (policyVersionChanged || authoritativeStatusChanged || changedEvidence)) { + [resultRow, contractRow] = await Promise.all([ + db.select().from(nativeRunResults).where(and( + eq(nativeRunResults.id, assessment.resultId), + eq(nativeRunResults.companyId, row.companyId), + eq(nativeRunResults.issueId, row.issueId), + eq(nativeRunResults.runId, row.runId), + )).limit(1).then((entries) => entries[0] ?? null), + db.select().from(completionContracts).where(and( + eq(completionContracts.id, assessment.contractId), + eq(completionContracts.companyId, row.companyId), + eq(completionContracts.issueId, row.issueId), + )).limit(1).then((entries) => entries[0] ?? null), + ]); + if (!resultRow || !contractRow) throw new Error("native_reconciliation_evidence_binding_missing"); + reassessment = await classifyNativeEvidence({ + db, + companyId: row.companyId, + issueId: row.issueId, + runId: row.runId, + contract: record(contractRow.contractJson), + result: record(record(resultRow.resultJson).result), + }); + } + const previousAssessment = record(assessment?.assessmentJson); + const newEvidenceSatisfiesContract = !!changedEvidence + && previousAssessment.allCriteriaSatisfied !== true + && reassessment?.allCriteriaSatisfied === true + && reassessment.verificationPassed === true; + const facts: NativeReconciliationFacts = policyVersionChanged + ? { policyVersionChanged: true } + : newEvidenceSatisfiesContract + ? { newEvidenceSatisfiesContract: true } + : authoritativeStatusChanged + ? { authoritativeStatusChanged: true } + : {}; + if (Object.keys(facts).length > 0) { + if (!assessment || !reassessment || !resultRow || !contractRow) { + throw new Error("native_reconciliation_reassessment_missing"); + } + const reassessmentRow = await recordNativeWorkAssessment({ + db, + companyId: row.companyId, + issueId: row.issueId, + runId: row.runId, + turnId: resultRow.turnId, + contractId: contractRow.id, + contractCanonicalSha256: contractRow.canonicalSha256, + resultId: resultRow.id, + resultCanonicalSha256: resultRow.canonicalSha256, + priorIssueStatus: row.issueStatus, + priorStatusVersion: Number(row.issueStatusVersion), + priorDecisionId: row.issueDecisionId, + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + assessment: reassessment, + supersedesAssessmentId: assessment.id, + }); + const decision = resolveNativeReconciliationStatus({ + facts, + priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, + agentId: row.agentId, + }); + const committed = await commitNativeStatusDecision({ + db, + companyId: row.companyId, + issueId: row.issueId, + runId: row.runId, + assessmentId: reassessmentRow.id, + priorStatus: row.issueStatus, + priorStatusVersion: Number(row.issueStatusVersion), + priorDecisionId: row.issueDecisionId, + decision, + allowSupersedingCommittedDecision: true, + }); + results.push({ + runId: row.runId, + phase: "committed", + reconciliationAction: decision.reasonCode, + decisionId: committed.decision.id, + reconciliationDecision: decision, + }); + continue; + } + } const barrier = await db.select({ status: workspaceOperations.status }) .from(workspaceOperations) .where(and( @@ -349,7 +508,48 @@ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) { .orderBy(desc(workspaceOperations.createdAt)) .limit(1) .then((entries) => entries[0] ?? null); - if (!barrier || !["succeeded", "failed"].includes(barrier.status)) continue; + if (!barrier || barrier.status !== "succeeded") { + const recoveryDecision = resolveNativeReconciliationStatus({ + facts: { workspaceOperationPending: true }, + priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, + agentId: row.agentId, + }); + if (!recoveryDecision.effects.some((effect) => effect.kind === "resume_workspace_operation")) { + throw new Error("native_reconciliation_workspace_resume_policy_missing"); + } + const operation = await resumeNativeWorkspaceFinalization({ db, runId: row.runId }); + const workspaceFinalizeStatus = operation.status === "succeeded" ? "succeeded" : "failed"; + try { + const finalized = await finalizeNativeRun({ + db, + runId: row.runId, + workspaceFinalizeStatus, + projectRunStatus: true, + }); + results.push({ + ...finalized, + reconciliationAction: "resume_workspace_operation" as const, + workspaceOperationId: operation.id, + workspaceFinalizeStatus, + reconciliationDecision: recoveryDecision, + }); + } catch (error) { + const failure = await recordNativeFinalizationFailure({ + db, + runId: row.runId, + error, + projectRunStatus: true, + }); + results.push({ + ...failure, + reconciliationAction: "resume_workspace_operation" as const, + workspaceOperationId: operation.id, + workspaceFinalizeStatus, + reconciliationDecision: recoveryDecision, + }); + } + continue; + } const replayDecision = resolveNativeReconciliationStatus({ facts: { canonicalReplay: true }, priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index 60ffccdddf4..c2374fbfd49 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -1,8 +1,10 @@ import type { Db } from "@paperclipai/db"; import type { NativeInteractionResponseEnvelope } from "@paperclipai/paperclip-runner"; -import { and, eq, inArray } from "drizzle-orm"; -import { heartbeatRuns, issues, issueThreadInteractions, nativeRunFinalizations } from "@paperclipai/db"; +import { and, asc, eq, inArray } from "drizzle-orm"; +import { agents, heartbeatRuns, issues, issueThreadInteractions, nativeRunFinalizations } from "@paperclipai/db"; +import { getAgentWorkEligibility } from "@paperclipai/shared"; import { issueThreadInteractionService } from "../issue-thread-interactions.js"; +import { commitNativeStatusDecision } from "./status-decision-committer.js"; import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, @@ -319,8 +321,132 @@ export type NativeAttentionFacts = { summary: string; budgetExhausted?: boolean; governanceGate?: NativeGovernanceGate | null; + resolvedTargetAgentId?: string | null; }; +export type NativeAttentionCandidate = { + route: NativeAttentionFacts["route"]; + summary: string; + responseState?: NativeAttentionFacts["responseState"]; + targetAgentId?: string | null; + budgetExhausted?: boolean; + governanceGate?: NativeGovernanceGate | null; + targetInteractionId?: string; + canonicalRequestId?: string; +}; + +/** + * Canonical production ingress for native attention. It resolves the run and + * issue binding, derives company scope, chooses an eligible in-company + * delegate, and commits (or explicitly audit-materializes) the resulting + * authority decision. + */ +export async function routeNativeAttention(input: { + db: Db; + runId: string; + candidate: NativeAttentionCandidate; +}) { + const binding = await input.db.select({ + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + runtimeMode: heartbeatRuns.runtimeMode, + issueId: nativeRunFinalizations.issueId, + assessmentId: nativeRunFinalizations.assessmentId, + issueStatus: issues.status, + issueStatusVersion: issues.statusVersion, + issueDecisionId: issues.lastStatusDecisionId, + }).from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .innerJoin(issues, and( + eq(issues.id, nativeRunFinalizations.issueId), + eq(issues.companyId, heartbeatRuns.companyId), + )) + .where(eq(heartbeatRuns.id, input.runId)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!binding || binding.runtimeMode !== "native") { + throw new NativeInteractionBridgeError("native_attention_binding_mismatch", "Native attention run binding not found"); + } + + const companyAgents = await input.db.select({ + id: agents.id, + companyId: agents.companyId, + name: agents.name, + status: agents.status, + reportsTo: agents.reportsTo, + }).from(agents).where(eq(agents.companyId, binding.companyId)).orderBy(asc(agents.createdAt), asc(agents.id)); + const targetSuggestion = input.candidate.targetAgentId + ? await input.db.select({ id: agents.id, companyId: agents.companyId }) + .from(agents).where(eq(agents.id, input.candidate.targetAgentId)).limit(1) + .then((rows) => rows[0] ?? null) + : null; + const companyScopeValid = !input.candidate.targetAgentId + || targetSuggestion?.companyId === binding.companyId; + let resolvedTargetAgentId: string | null = null; + if (companyScopeValid && input.candidate.route === "agent") { + const eligible = companyAgents.filter((candidate) => + candidate.id !== binding.agentId + && getAgentWorkEligibility({ agent: candidate, agents: companyAgents }).invokable + ); + const requested = eligible.find((candidate) => candidate.id === input.candidate.targetAgentId); + resolvedTargetAgentId = requested?.id ?? eligible[0]?.id ?? null; + } + + const facts: NativeAttentionFacts = { + companyScopeValid, + responseState: input.candidate.responseState ?? "none", + route: input.candidate.route === "agent" && !resolvedTargetAgentId + ? "recovery" + : input.candidate.route, + summary: input.candidate.summary, + budgetExhausted: input.candidate.budgetExhausted, + governanceGate: input.candidate.governanceGate, + resolvedTargetAgentId, + }; + const decision = resolveNativeAttentionStatus({ + facts, + priorIssueStatus: binding.issueStatus as NativeAuthoritativeIssueStatus, + agentId: binding.agentId, + }); + const auditOnly = !companyScopeValid + || decision.reasonCode === "attention_duplicate_suppressed"; + if (auditOnly) { + const targets = await applyNativeAttentionStatusDecision({ + db: input.db, + companyId: binding.companyId, + issueId: binding.issueId, + runId: input.runId, + decision, + targetInteractionId: input.candidate.targetInteractionId, + canonicalRequestId: input.candidate.canonicalRequestId, + }); + return { decision, decisionId: null, targets, resolvedTargetAgentId }; + } + if (!binding.assessmentId) { + throw new NativeInteractionBridgeError( + "native_attention_assessment_missing", + "Native attention requires the server-owned work assessment for this result", + ); + } + const committed = await commitNativeStatusDecision({ + db: input.db, + companyId: binding.companyId, + issueId: binding.issueId, + runId: input.runId, + assessmentId: binding.assessmentId, + priorStatus: binding.issueStatus, + priorStatusVersion: Number(binding.issueStatusVersion), + priorDecisionId: binding.issueDecisionId, + decision, + }); + return { + decision, + decisionId: committed.decision.id, + targets: [], + resolvedTargetAgentId, + }; +} + /** Server-owned attention resolution from canonical routing facts. */ export function resolveNativeAttentionStatus(input: { facts: NativeAttentionFacts; @@ -395,11 +521,19 @@ export function resolveNativeAttentionStatus(input: { return continuation("attention_resolved_from_context"); } if (input.facts.route === "agent") { + if (!input.facts.resolvedTargetAgentId) { + return preserve("attention_budget_exhausted", [{ + kind: "record_finalization_error", + cause: "attention_no_valid_route", + nextAction: "Assign a same-company eligible delegate.", + agentId: input.agentId, + }]); + } const decision = continuation("attention_routed_to_agent"); return { ...decision, effects: [ - { kind: "create_delegated_issue", agentId: input.agentId, summary: input.facts.summary }, + { kind: "create_delegated_issue", agentId: input.facts.resolvedTargetAgentId, summary: input.facts.summary }, ...decision.effects, ], }; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index a0facd3cfb9..36b6348759d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -15,6 +15,8 @@ import { and, eq, sql } from "drizzle-orm"; import { heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; +import { persistActivity, publishActivity } from "../activity-log.js"; +import { commitNativeStatusDecision } from "./status-decision-committer.js"; import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, @@ -48,58 +50,146 @@ export function cancelNativeSession( runId: string, reason: string, options: { db: Db; scope?: "turn" | "run" | "issue"; replacementAccepted?: boolean }, -): Promise<{ dispatched: boolean; decision: NativeStatusDecision | null }>; +): Promise<{ dispatched: boolean; decision: NativeStatusDecision | null; decisionId: string | null; auditId: string | null }>; export async function cancelNativeSession( runId: string, reason: string, options?: { db: Db; scope?: "turn" | "run" | "issue"; replacementAccepted?: boolean }, -): Promise { +): Promise { let decision: NativeStatusDecision | null = null; + let decisionContext: { + companyId: string; + issueId: string; + assessmentId: string | null; + priorStatus: string; + priorStatusVersion: number; + priorDecisionId: string | null; + agentId: string; + } | null = null; if (options) { const run = await options.db.select({ agentId: heartbeatRuns.agentId, companyId: heartbeatRuns.companyId, contextSnapshot: heartbeatRuns.contextSnapshot, + runtimeMode: heartbeatRuns.runtimeMode, }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1).then((rows) => rows[0] ?? null); const context = record(run?.contextSnapshot); const issueId = typeof context.issueId === "string" ? context.issueId : typeof context.taskId === "string" ? context.taskId : null; const issue = run && issueId - ? await options.db.select({ status: issues.status }).from(issues).where(and( + ? await options.db.select({ + status: issues.status, + statusVersion: issues.statusVersion, + lastStatusDecisionId: issues.lastStatusDecisionId, + }).from(issues).where(and( eq(issues.id, issueId), eq(issues.companyId, run.companyId), )).limit(1).then((rows) => rows[0] ?? null) : null; - if (run && issue) { + if (run && issue && issueId && run.runtimeMode === "native") { decision = resolveNativeCancellationStatus({ scope: options.scope ?? "run", priorIssueStatus: issue.status as NativeAuthoritativeIssueStatus, agentId: run.agentId, replacementAccepted: options.replacementAccepted, }); - if (options.replacementAccepted && decision.effects.some((effect) => effect.kind === "accept_replacement_turn")) { - await options.db.update(heartbeatRuns).set({ - status: "running", - continuationAttempt: sql`${heartbeatRuns.continuationAttempt} + 1`, - nextAction: "Accept a replacement native turn on the existing run.", - updatedAt: new Date(), - }).where(eq(heartbeatRuns.id, runId)); - } + const coordinator = await options.db.select({ + assessmentId: nativeRunFinalizations.assessmentId, + }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.companyId, run.companyId), + eq(nativeRunFinalizations.issueId, issueId), + )).limit(1).then((rows) => rows[0] ?? null); + decisionContext = { + companyId: run.companyId, + issueId, + assessmentId: coordinator?.assessmentId ?? null, + priorStatus: issue.status, + priorStatusVersion: Number(issue.statusVersion), + priorDecisionId: issue.lastStatusDecisionId, + agentId: run.agentId, + }; } } const active = activeNativeSessions.get(runId); - if (!active) return options ? { dispatched: false, decision } : false; - if (active.cancelRequested) return options ? { dispatched: true, decision } : true; - active.cancelRequested = true; - try { - if (active.session.cancel) await active.session.cancel({ reason }); - else if (active.session.interrupt) await active.session.interrupt({ reason }); - } catch (error) { - active.cancelRequested = false; - throw error; + let dispatched = false; + if (active) { + dispatched = true; + if (!active.cancelRequested) { + active.cancelRequested = true; + try { + if (active.session.cancel) await active.session.cancel({ reason }); + else if (active.session.interrupt) await active.session.interrupt({ reason }); + } catch (error) { + active.cancelRequested = false; + throw error; + } + } } - return options ? { dispatched: true, decision } : true; + if (!options) return dispatched; + + let decisionId: string | null = null; + let auditId: string | null = null; + if (decision && decisionContext) { + if (decisionContext.assessmentId && decision.reasonCode !== null) { + const committed = await commitNativeStatusDecision({ + db: options.db, + companyId: decisionContext.companyId, + issueId: decisionContext.issueId, + runId, + assessmentId: decisionContext.assessmentId, + priorStatus: decisionContext.priorStatus, + priorStatusVersion: decisionContext.priorStatusVersion, + priorDecisionId: decisionContext.priorDecisionId, + decision, + }); + decisionId = committed.decision.id; + } else if (options.replacementAccepted && decision.effects.some((effect) => effect.kind === "accept_replacement_turn")) { + await options.db.update(heartbeatRuns).set({ + status: "running", + continuationAttempt: sql`${heartbeatRuns.continuationAttempt} + 1`, + nextAction: "Accept a replacement native turn on the existing run.", + updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, runId)); + } + const currentRun = await options.db.select({ resultJson: heartbeatRuns.resultJson }) + .from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)).limit(1).then((rows) => rows[0] ?? null); + await options.db.update(heartbeatRuns).set({ + resultJson: { + ...record(currentRun?.resultJson), + nativeCancellation: { + scope: options.scope ?? "run", + reasonCode: decision.reasonCode, + effects: decision.effects.map((effect) => effect.kind), + dispatched, + decisionId, + }, + }, + updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, runId)); + const activity = await persistActivity(options.db, { + companyId: decisionContext.companyId, + actorType: "system", + actorId: "native-session-cancellation", + action: decisionId ? "issue.status_decision_recorded" : "native.cancellation_audited", + entityType: "heartbeat_run", + entityId: runId, + agentId: decisionContext.agentId, + runId, + issueId: decisionContext.issueId, + details: { + scope: options.scope ?? "run", + reasonCode: decision.reasonCode, + effects: decision.effects.map((effect) => effect.kind), + dispatched, + decisionId, + }, + }); + auditId = activity.activity?.id ?? null; + publishActivity(activity.publication); + } + return { dispatched, decision, decisionId, auditId }; } /** Authenticated cancellation scope projected through the shared arbiter. */ diff --git a/server/src/services/native-runtime/native-workspace-finalizer.ts b/server/src/services/native-runtime/native-workspace-finalizer.ts new file mode 100644 index 00000000000..49e87e7795a --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-finalizer.ts @@ -0,0 +1,133 @@ +import fs from "node:fs/promises"; +import { and, desc, eq } from "drizzle-orm"; +import type { Db } from "@paperclipai/db"; +import { + executionWorkspaces, + heartbeatRuns, + issues, + nativeRunFinalizations, + workspaceOperations, +} from "@paperclipai/db"; +import { workspaceOperationService } from "../workspace-operations.js"; +import { inspectManagedGitWorktreeBranch } from "../workspace-runtime.js"; + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? value as Record + : {}; +} + +function readString(value: unknown) { + return typeof value === "string" && value.trim().length > 0 ? value.trim() : null; +} + +/** + * Resume the real workspace-finalization action for a result-bearing native + * run. This service observes the workspace; it never fabricates a successful + * marker. The caller decides whether a failed observation is retryable. + */ +export async function resumeNativeWorkspaceFinalization(input: { + db: Db; + runId: string; +}) { + const bound = await input.db.select({ + companyId: heartbeatRuns.companyId, + runtimeMode: heartbeatRuns.runtimeMode, + runnerProfileJson: heartbeatRuns.runnerProfileJson, + issueId: nativeRunFinalizations.issueId, + resultId: nativeRunFinalizations.resultId, + issueWorkspaceId: issues.executionWorkspaceId, + }) + .from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .innerJoin(issues, and( + eq(issues.id, nativeRunFinalizations.issueId), + eq(issues.companyId, heartbeatRuns.companyId), + )) + .where(eq(heartbeatRuns.id, input.runId)) + .limit(1) + .then((rows) => rows[0] ?? null); + if (!bound || bound.runtimeMode !== "native" || !bound.resultId) { + throw new Error("native_workspace_finalization_binding_missing"); + } + + const previous = await input.db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, bound.companyId), + eq(workspaceOperations.heartbeatRunId, input.runId), + eq(workspaceOperations.phase, "workspace_finalize"), + )).orderBy(desc(workspaceOperations.createdAt)).limit(1).then((rows) => rows[0] ?? null); + + const persistedInput = record(record(bound.runnerProfileJson).nativeExecutionInput); + const binding = record(persistedInput.binding); + const workspaceId = previous?.executionWorkspaceId + ?? bound.issueWorkspaceId + ?? readString(binding.executionWorkspaceId); + const workspace = workspaceId + ? await input.db.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.id, workspaceId), + eq(executionWorkspaces.companyId, bound.companyId), + )).limit(1).then((rows) => rows[0] ?? null) + : null; + const cwd = workspace?.providerRef ?? workspace?.cwd ?? previous?.cwd ?? null; + + const recorder = workspaceOperationService(input.db).createRecorder({ + companyId: bound.companyId, + heartbeatRunId: input.runId, + executionWorkspaceId: workspace?.id ?? workspaceId, + issueId: bound.issueId, + }); + return recorder.recordOperation({ + phase: "workspace_finalize", + cwd, + metadata: { + resumedFromOperationId: previous?.id ?? null, + owningService: "native_workspace_finalizer", + observation: workspace?.strategyType === "git_worktree" + ? "managed_git_worktree_branch" + : "workspace_directory", + }, + run: async () => { + if (!cwd) { + return { + status: "failed", + exitCode: 1, + stderr: "Native workspace finalization cannot resolve the persisted workspace path.\n", + }; + } + const isDirectory = await fs.stat(cwd).then((stat) => stat.isDirectory()).catch(() => false); + if (!isDirectory) { + return { + status: "failed", + exitCode: 1, + stderr: `Native workspace finalization could not access workspace directory ${cwd}.\n`, + }; + } + if (workspace?.strategyType === "git_worktree") { + const inspection = await inspectManagedGitWorktreeBranch({ + worktreePath: cwd, + expectedBranchName: workspace.branchName, + }); + if (!inspection.valid) { + return { + status: "failed", + exitCode: 1, + stderr: `Managed git worktree validation failed: ${inspection.reason ?? "unknown failure"}.\n`, + metadata: { managedGitWorktreeBranch: inspection }, + }; + } + return { + status: "succeeded", + exitCode: 0, + system: "Native workspace finalization validated the managed git worktree.\n", + metadata: { managedGitWorktreeBranch: inspection }, + }; + } + return { + status: "succeeded", + exitCode: 0, + system: "Native workspace finalization validated the persisted workspace directory.\n", + metadata: { observedWorkspacePath: cwd }, + }; + }, + }); +} diff --git a/server/src/services/native-runtime/runtime-mode.ts b/server/src/services/native-runtime/runtime-mode.ts index 86b2e72bdde..01e4c99cd39 100644 --- a/server/src/services/native-runtime/runtime-mode.ts +++ b/server/src/services/native-runtime/runtime-mode.ts @@ -45,6 +45,13 @@ export function resolveNativeRuntimeMode(input: { }): NativeRuntimeResolution { const nativeRunner = record(record(input.runtimeConfig).nativeRunner); const mode = nativeRunner.mode; + if (nativeRunner.disabledByNativeKillSwitch === true) { + return { + kind: "legacy", + resolverVersion: NATIVE_RUNTIME_RESOLVER_VERSION, + reason: "native_kill_switch_persisted", + }; + } if (!input.enabled) { const compatibility = resolveNativeCompatibilityStatus({ facts: { shadowApplicationDisabled: true }, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index c5770abbb11..5f2a9f04228 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -3,6 +3,7 @@ import type { Db } from "@paperclipai/db"; import type { CreateIssueThreadInteraction } from "@paperclipai/shared"; import { agentWakeupRequests, + agents, approvals, completionContracts, heartbeatRuns, @@ -13,6 +14,7 @@ import { nativeRunFinalizations, statusDecisionEffects, statusDecisions, + workAssessments, workspaceOperations, } from "@paperclipai/db"; import { @@ -24,6 +26,7 @@ import { nativeSha256 } from "./canonical.js"; import { issueService } from "../issues.js"; import { issueThreadInteractionService } from "../issue-thread-interactions.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; +import { agentService } from "../agents.js"; import { buildIssueBlockersResolvedWakeIdempotencyKey } from "../issue-dependency-wakeups.js"; import { persistActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; @@ -44,6 +47,16 @@ export async function dispatchPendingNativeStatusEffects(input: { runId: string; }) { return input.db.transaction(async (tx) => { + const coordinator = await tx.select({ + decisionId: nativeRunFinalizations.decisionId, + }).from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.runId, input.runId), + eq(nativeRunFinalizations.companyId, input.companyId), + eq(nativeRunFinalizations.issueId, input.issueId), + )).for("update").limit(1).then((rows) => rows[0] ?? null); + if (!coordinator || coordinator.decisionId !== input.decisionId) { + throw new Error("native_pending_effect_decision_binding_mismatch"); + } const pending = await tx.select({ id: statusDecisionEffects.id, effectKind: statusDecisionEffects.effectKind, @@ -93,7 +106,7 @@ export type NativeStatusCommitFailpoint = | "recovery_materialization" | "status_projection"; -type MaterializedEffect = { +export type NativeMaterializedStatusEffect = { effectKind: string; targetType: string; targetId: string | null; @@ -125,6 +138,11 @@ async function assertPendingEffectTarget(tx: Db, input: { eq(agentWakeupRequests.id, input.targetId), eq(agentWakeupRequests.companyId, input.companyId), )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "agent") { + found = await tx.select({ id: agents.id }).from(agents).where(and( + eq(agents.id, input.targetId), + eq(agents.companyId, input.companyId), + )).limit(1).then((rows) => rows.length === 1); } else if (input.targetType === "issue_thread_interaction" || input.targetType === "interaction") { found = await tx.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(and( eq(issueThreadInteractions.id, input.targetId), @@ -306,7 +324,8 @@ async function materializeDecisionEffect(input: { decisionId: string; effect: NativeStatusEffect; failpoint?: NativeStatusCommitFailpoint; -}): Promise { + preMaterializedEffects?: ReadonlyMap; +}): Promise { const { effect } = input; if (effect.kind === "create_interaction") { failAt("governance_materialization", input.failpoint); @@ -396,8 +415,7 @@ async function materializeDecisionEffect(input: { }; } if (effect.kind === "create_delegated_issue") { - const delegated = await input.tx.insert(issues).values({ - companyId: input.companyId, + const delegated = await issueService(input.tx).create(input.companyId, { projectId: input.issue.projectId, projectWorkspaceId: input.issue.projectWorkspaceId, goalId: input.issue.goalId, @@ -408,7 +426,9 @@ async function materializeDecisionEffect(input: { assigneeAgentId: effect.agentId, createdByAgentId: input.issue.assigneeAgentId, workMode: "standard", - }).returning({ id: issues.id }).then((rows) => rows[0] ?? null); + idempotencyKey: `native-delegation:${input.decisionId}`, + inheritExecutionWorkspaceFromIssueId: input.issue.id, + }); if (!delegated) throw new Error("native_delegated_issue_not_persisted"); return { effectKind: effect.kind, @@ -666,6 +686,27 @@ async function materializeDecisionEffect(input: { } if (effect.kind === "append_superseding_assessment") { if (!input.issue.lastStatusDecisionId) throw new Error("native_superseded_decision_missing"); + const lineage = await input.tx.select({ + currentAssessmentId: statusDecisions.assessmentId, + }).from(statusDecisions).where(and( + eq(statusDecisions.id, input.decisionId), + eq(statusDecisions.companyId, input.companyId), + )).limit(1).then((rows) => rows[0] ?? null); + const prior = await input.tx.select({ + assessmentId: statusDecisions.assessmentId, + }).from(statusDecisions).where(and( + eq(statusDecisions.id, input.issue.lastStatusDecisionId), + eq(statusDecisions.companyId, input.companyId), + )).limit(1).then((rows) => rows[0] ?? null); + if (!lineage || !prior) throw new Error("native_superseded_assessment_missing"); + const [assessment] = await input.tx.update(workAssessments).set({ + supersedesAssessmentId: prior.assessmentId, + }).where(and( + eq(workAssessments.id, lineage.currentAssessmentId), + eq(workAssessments.companyId, input.companyId), + eq(workAssessments.issueId, input.issue.id), + )).returning({ id: workAssessments.id }); + if (!assessment) throw new Error("native_superseding_assessment_not_linked"); const [decision] = await input.tx.update(statusDecisions).set({ supersedesDecisionId: input.issue.lastStatusDecisionId, }).where(and( @@ -677,7 +718,11 @@ async function materializeDecisionEffect(input: { effectKind: effect.kind, targetType: "status_decision", targetId: decision.id, - payload: { supersedesDecisionId: input.issue.lastStatusDecisionId }, + payload: { + supersedesDecisionId: input.issue.lastStatusDecisionId, + assessmentId: assessment.id, + supersedesAssessmentId: prior.assessmentId, + }, }; } if (effect.kind === "dispatch_pending_effect") { @@ -711,13 +756,11 @@ async function materializeDecisionEffect(input: { }; } if (effect.kind === "increment_status_version") { - const [updated] = await input.tx.update(issues).set({ - statusVersion: sql`${issues.statusVersion} + 1`, - updatedAt: new Date(), - }).where(and( - eq(issues.id, input.issue.id), - eq(issues.companyId, input.companyId), - )).returning({ id: issues.id, statusVersion: issues.statusVersion }); + const updated = await issueService(input.tx).update(input.issue.id, { + statusVersion: Number(input.issue.statusVersion) + 1, + actorAgentId: null, + actorUserId: null, + }, input.tx); if (!updated) throw new Error("native_status_version_not_incremented"); return { effectKind: effect.kind, @@ -839,42 +882,54 @@ async function materializeDecisionEffect(input: { return { effectKind: effect.kind, targetType: "heartbeat_run", targetId, payload: { runtimeMode: "native" } }; } if (effect.kind === "stop_new_native_dispatch") { - const targetId = await updateRunEffectState({ - tx: input.tx, - companyId: input.companyId, - runId: input.runId, - key: "nativeDispatchAfterRun", - value: "legacy", - runnerProfile: true, + const run = await input.tx.select({ agentId: heartbeatRuns.agentId }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), + )).limit(1).then((rows) => rows[0] ?? null); + if (!run) throw new Error("native_kill_switch_run_missing"); + const agent = await agentService(input.tx).getById(run.agentId); + if (!agent || agent.companyId !== input.companyId) throw new Error("native_kill_switch_agent_missing"); + const runtimeConfig = record(agent.runtimeConfig); + const nativeRunner = record(runtimeConfig.nativeRunner); + const updated = await agentService(input.tx).update(run.agentId, { + runtimeConfig: { + ...runtimeConfig, + nativeRunner: { + ...nativeRunner, + mode: "legacy", + disabledByNativeKillSwitch: true, + }, + }, + }, { + recordRevision: { + source: "native_kill_switch", + createdByAgentId: null, + createdByUserId: null, + }, }); - return { effectKind: effect.kind, targetType: "heartbeat_run", targetId, payload: { nextDispatchMode: "legacy" } }; - } - if (effect.kind === "resume_workspace_operation") { - const previous = await input.tx.select().from(workspaceOperations).where(and( - eq(workspaceOperations.companyId, input.companyId), - eq(workspaceOperations.heartbeatRunId, input.runId), - eq(workspaceOperations.phase, "workspace_finalize"), - )).orderBy(desc(workspaceOperations.createdAt)).limit(1).then((rows) => rows[0] ?? null); - if (!previous) throw new Error("native_workspace_operation_missing"); - const [resumed] = await input.tx.insert(workspaceOperations).values({ - companyId: input.companyId, - executionWorkspaceId: previous.executionWorkspaceId, - heartbeatRunId: input.runId, - issueId: input.issue.id, - phase: "workspace_finalize", - status: "succeeded", - exitCode: 0, - metadata: { resumedFromOperationId: previous.id, nativeDecisionId: input.decisionId }, - finishedAt: new Date(), - }).returning({ id: workspaceOperations.id }); - if (!resumed) throw new Error("native_workspace_operation_not_resumed"); + if (!updated) throw new Error("native_kill_switch_not_persisted"); return { effectKind: effect.kind, - targetType: "workspace_operation", - targetId: resumed.id, - payload: { resumedFromOperationId: previous.id }, + targetType: "agent", + targetId: updated.id, + payload: { nextDispatchMode: "legacy", runtimeConfigOwner: updated.id }, }; } + if (effect.kind === "resume_workspace_operation") { + const resumed = input.preMaterializedEffects?.get(effect.kind); + if (!resumed || resumed.targetType !== "workspace_operation" || !resumed.targetId) { + throw new Error("native_workspace_operation_not_executed"); + } + const operation = await input.tx.select({ status: workspaceOperations.status }).from(workspaceOperations).where(and( + eq(workspaceOperations.id, resumed.targetId), + eq(workspaceOperations.companyId, input.companyId), + eq(workspaceOperations.heartbeatRunId, input.runId), + eq(workspaceOperations.issueId, input.issue.id), + eq(workspaceOperations.phase, "workspace_finalize"), + )).limit(1).then((rows) => rows[0] ?? null); + if (operation?.status !== "succeeded") throw new Error("native_workspace_operation_not_succeeded"); + return resumed; + } if (effect.kind === "record_stale_response") { const interaction = await input.tx.select({ id: issueThreadInteractions.id }) .from(issueThreadInteractions).where(and( @@ -964,6 +1019,8 @@ export async function commitNativeStatusDecision(input: { priorDecisionId: string | null; decision: NativeStatusDecision; failpoint?: NativeStatusCommitFailpoint; + preMaterializedEffects?: NativeMaterializedStatusEffect[]; + allowSupersedingCommittedDecision?: boolean; }) { if (input.decision.reasonCode === null) { throw new Error("native_status_reason_code_required"); @@ -983,13 +1040,22 @@ export async function commitNativeStatusDecision(input: { )).for("update").limit(1).then((rows) => rows[0] ?? null); if (!issue) throw new NativeStatusRaceError(); if (coordinator.phase === "committed" && coordinator.decisionId) { - const existingDecision = await tx.select().from(statusDecisions).where(and( - eq(statusDecisions.id, coordinator.decisionId), - eq(statusDecisions.companyId, input.companyId), - eq(statusDecisions.issueId, input.issueId), - )).limit(1).then((rows) => rows[0] ?? null); - if (!existingDecision) throw new Error("native_committed_decision_missing"); - return { decision: existingDecision, issue, replayed: true }; + if (input.allowSupersedingCommittedDecision) { + if ( + coordinator.decisionId !== input.priorDecisionId + || coordinator.assessmentId === input.assessmentId + ) { + throw new NativeStatusRaceError(); + } + } else { + const existingDecision = await tx.select().from(statusDecisions).where(and( + eq(statusDecisions.id, coordinator.decisionId), + eq(statusDecisions.companyId, input.companyId), + eq(statusDecisions.issueId, input.issueId), + )).limit(1).then((rows) => rows[0] ?? null); + if (!existingDecision) throw new Error("native_committed_decision_missing"); + return { decision: existingDecision, issue, replayed: true }; + } } if ( issue.status !== input.priorStatus @@ -1038,7 +1104,10 @@ export async function commitNativeStatusDecision(input: { } if (!decisionRow) throw new Error("native_status_decision_not_persisted"); - const materialized: MaterializedEffect[] = []; + const materialized: NativeMaterializedStatusEffect[] = []; + const preMaterializedEffects = new Map( + (input.preMaterializedEffects ?? []).map((effect) => [effect.effectKind, effect]), + ); for (const effect of input.decision.effects) { materialized.push(await materializeDecisionEffect({ tx: tx as unknown as Db, @@ -1048,6 +1117,7 @@ export async function commitNativeStatusDecision(input: { decisionId: decisionRow.id, effect, failpoint: input.failpoint, + preMaterializedEffects, })); }