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 43ce99b52d2..1847ad651d8 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 @@ -1,6 +1,6 @@ # Phase 6 Thin Paperclip Adapter Boundary -Status: design approved; implementation remediated; CTO conformance re-review pending +Status: design approved; remediation 4 implemented; CTO conformance re-review pending Date: 2026-08-09 Decision scope: the first feature-flagged Paperclip native run, including the normative native finalizer and server-owned Section 18 status arbitration @@ -797,6 +797,32 @@ that consumer did not execute or returned different semantics. No hand-written test policy table supplies observed outcomes and no row is satisfied by a coverage-label join to a generic observation. +### Remediation 4 live-consumer and effect-materialization amendment (2026-08-09) + +The fixture-keyed scenario arbiter is removed. Finalization now calls the same +fact-based arbiter as the heartbeat finalizer, while attention, cancellation, +reconciliation, compatibility, and migration decisions accept canonical facts +and each has a non-test production caller. The corpus constructs those durable +facts, invokes the production consumer, and uses that return plus persisted +issue, run, decision, interaction, wake, recovery, workspace-operation, and +contract rows as its observation. + +`StatusDecisionCommitter` handles every `NativeStatusEffect` explicitly. Each +case creates or changes its named target before its delivered ledger row is +written. Reconciliation acknowledges a pending effect only after its existing +company- and issue-bound target is verified. An unknown effect or target type +throws inside the transaction, leaving the decision, effect ledger, issue +status/version, and finalization coordinator unchanged. Corpus replay calls the +committer twice and requires one decision identity and exactly one delivery +attempt per target; audit-only attention and replacement-turn paths likewise +assert their real target state. + +The 52-fixture/70-row proof therefore depends on runtime-reachable consumer +returns and materialized state. Removing a production consumer invocation, +changing its facts or decision, suppressing its target mutation, duplicating a +delivery, or restoring a synthetic `issue_checkout` fallback fails the focused +database gate. + ## 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 95311b92c0e..8d31da4362e 100644 --- a/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md +++ b/packages/paperclip-runner/docs/phase-06-thin-paperclip-adapter.md @@ -61,7 +61,9 @@ terminal state succeeded, every criterion and verification has accepted durable evidence, and no blocking work remains. It marks `blocked` only for a task-wide first-class blocker with a named owner and action. Review, retry, continuation, blocker, and recovery effects materialize atomically with the -status-version CAS and audit record. +status-version CAS and audit record. The committer has no catch-all delivery: +all native effect kinds must create or mutate their named target, and unknown +effects fail the transaction closed. ## Recovery and safety @@ -104,7 +106,17 @@ committer, reconciliation, compatibility, or migration consumer and fails if that consumer did not execute or returned different semantics. A per-fixture mutation check independently changes every expected field and proves the comparison rejects it; there is no test-owned policy table supplying observed -semantics. +semantics. Finalizer observations come from the live fact-based arbiter rather +than a fixture-state switch. Every delivered effect is joined to an actual +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. 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 ec38aa846e0..41be67ec18b 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-one targeted tests check company binding, duplicate/gap replay, +port. Forty-two 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, @@ -76,7 +76,11 @@ 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. The recovery case uses a scripted provider only at the backend +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 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 794e4bd493c..166104d7f62 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 @@ -4,7 +4,7 @@ title: Phase 6 thin Paperclip adapter verification description: Targeted package, database-port, finalization, security-boundary, feature-selection, and legacy fallback evidence. tags: [native-runner, phase-6, verification, paperclip, replay, finalization] status: stable -generated: { by: openai/gpt-5.6, at: 2026-08-09T05:20:00Z } +generated: { by: openai/gpt-5.6, at: 2026-08-09T05:52:00Z } --- # Scope @@ -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-16877 remediation rerun. “Internal canary” means +The counts below are the PAP-16878 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 41 tests in thirteen files against +The complete acceptance gate below passed 42 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 @@ -76,13 +76,13 @@ The focused corpus also proved: - a completion decision writes immutable assessment/decision/effect records and an activity-log audit row; - a failed workspace barrier preserves the accepted semantic result, leaves the - issue status unchanged, and records a leased `retryable_failure` plus recovery - owner/action; + issue status/version unchanged, commits the fact-based preserve decision and + concrete recovery target, and records a leased `retryable_failure`; - six injected failures prove governance binding, interactions, wakes, blocker bindings, recovery, status, decisions, and effects roll back together before retry ownership is recorded. -The acceptance-matrix entry points passed as one 13-file, 41-test gate: +The acceptance-matrix entry points passed as one 13-file, 42-test gate: ```sh pnpm --filter @paperclipai/server exec vitest run \ @@ -113,7 +113,10 @@ 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. No test-owned policy table supplies an observation. The selected-task +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 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 cac040efef0..500b9c7fe7d 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 @@ -47,6 +47,16 @@ existing server authority and keep legacy execution as the default. 12. Project supported resolved interaction responses through the existing authorized issue-interaction service. Governed, unresolved, unsupported, cross-boundary, or self-approved requests fail closed without credentials. +13. Remove fixture-scenario arbitration from production. Finalization consumes + durable evidence facts, and attention, cancellation, reconciliation, + compatibility, and migration consumers are invoked from their live server + paths. +14. Handle every native status effect exhaustively. A delivered row is written + only after its named target exists or changes; pending delivery verifies the + original target, and unknown effects or target types roll back. +15. Prove at-most-once materialization by replaying each corpus decision and + requiring one decision identity, one delivery attempt, and the persisted + target state. # Evidence @@ -54,7 +64,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-one tests passed +- Phase 6 acceptance-matrix entry points: thirteen files, forty-two tests passed with zero skips, including the database-backed selected-task canary, six atomic-liveness failpoints, migration, sequencing, bounded recovery, and legacy compatibility. @@ -69,6 +79,10 @@ existing server authority and keep legacy execution as the default. count. All 70 matrix rows require their named consumer execution and compare that consumer's semantics. Per-fixture mutation checks reject changes to every expected field; no test-owned policy table supplies observations. +- All native effect observations resolve to actual interaction, wake, issue, + recovery, run, workspace-operation, contract, decision, or governance state. + Unknown effects fail before any decision, ledger, status-version, or + coordinator mutation. - 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 cb1c6095dcd..50dd739b4d1 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -19,10 +19,12 @@ import { statusDecisionEffects, statusDecisions, workAssessments, + workspaceOperations, } from "@paperclipai/db"; import type { NativeEvidenceAssessment } from "../services/native-runtime/evidence-classifier.js"; import { classifyNativeEvidence } from "../services/native-runtime/evidence-classifier.js"; import { + applyNativeAttentionStatusDecision, materializeNativeInteractionResponses, rejectUnsupportedNativeRuntimeRequest, resolveNativeAttentionStatus, @@ -41,7 +43,8 @@ import { } from "../services/native-runtime/runtime-mode.js"; import { arbitrateNativeStatus, - type NativeStatusAuthorityScenario, + type NativeStatusDecision, + type NativeStatusEffect, } from "../services/native-runtime/status-arbiter.js"; import { commitNativeStatusDecision, @@ -52,7 +55,10 @@ import { recordNativeFinalizationFailure, resolveNativeFinalizerStatus, } from "../services/native-runtime/native-run-finalizer.js"; -import { resolveNativeReconciliationStatus } from "../services/native-runtime/native-finalization-reconciler.js"; +import { + reconcileNativeFinalizations, + resolveNativeReconciliationStatus, +} from "../services/native-runtime/native-finalization-reconciler.js"; import { issueService } from "../services/issues.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -211,27 +217,21 @@ const completeEvidenceStates = new Set([ "cohort_policy_pinned", ]); -const scenarioStates = new Set([ - "mechanically_satisfied", "low_risk_policy_claim", "missing_required_test", - "no_durable_continuation", "named_reviewer_required", "governed_gate_pending", - "review_required", "safe_partial_parse", "valid_continuation", - "alternate_track_runnable", "task_wide_owner_action_bound", "context_answer_current", - "ordinary_domain_expertise", "intentional_human_judgment", "equivalent_attention_family", - "resolver_budget_exhausted", "partial_evidence_persisted", "claim_before_workspace_failure", - "turn_scope", "run_scope", "issue_scope_authorized", "new_evidence_satisfies_contract", - "dependency_now_done", "explicit_resume_capability", "transient_retry_then_success", - "replacement_turn_accepted", - "cross_company_target", "response_after_supersession", "interaction_expired", - "identical_result_before_ack", "reused_id_changed_material", "result_preserved", - "decision_committed_delivery_pending", "board_cancelled_before_cas", "new_policy_requires_review", - "shadow_application_disabled", "mixed_ledger", "authorized_writer_incremented_version", - "shadow_compute", "classified_native_legacy_divergence", "allowlisted_company_adapter_policy", - "cohort_policy_pinned", "kill_switch_during_active_native_run", -]); - const zeroDecisionStates = new Set([ "safe_partial_parse", "equivalent_attention_family", "cross_company_target", - "response_after_supersession", "reused_id_changed_material", + "response_after_supersession", "reused_id_changed_material", "decision_committed_delivery_pending", +]); + +const nativeStatusEffectKinds = new Set([ + "create_interaction", "bind_reviewer", "notify_owner", "enqueue_continuation", + "bind_blocker", "schedule_retry", "record_finalization_error", "release_run_resources", + "create_delegated_issue", "accept_replacement_turn", "cancel_continuations", + "append_superseding_assessment", "dispatch_pending_effect", "increment_status_version", + "schedule_reconciliation", "record_shadow_decision", "render_four_layers", + "materialize_contract", "record_mode_labeled_divergence", "record_mode_native", + "record_policy_version", "finish_as_native", "stop_new_native_dispatch", + "resume_workspace_operation", "record_expiry", "record_stale_response", + "link_canonical_request", "record_recovery", "release_checkout", ]); const supersedingDecisionStates = new Set([ @@ -263,6 +263,80 @@ function failpointFor(fixture: Fixture): NativeStatusCommitFailpoint | undefined return undefined; } +function attentionFactsFor(completionState: string, summary: string, governanceGate: { kind: "interaction"; id: string } | null) { + switch (completionState) { + case "alternate_track_runnable": + return { companyScopeValid: true, responseState: "none" as const, route: "alternate_track" as const, summary }; + case "context_answer_current": + return { companyScopeValid: true, responseState: "resolved" as const, route: "context" as const, summary }; + case "ordinary_domain_expertise": + return { companyScopeValid: true, responseState: "none" as const, route: "agent" as const, summary }; + case "intentional_human_judgment": + return { companyScopeValid: true, responseState: "none" as const, route: "human" as const, summary }; + case "equivalent_attention_family": + return { companyScopeValid: true, responseState: "none" as const, route: "duplicate" as const, summary }; + case "resolver_budget_exhausted": + return { companyScopeValid: true, responseState: "none" as const, route: "recovery" as const, summary, budgetExhausted: true }; + case "transient_retry_then_success": + return { companyScopeValid: true, responseState: "resolved" as const, route: "retry" as const, summary }; + case "cross_company_target": + return { companyScopeValid: false, responseState: "none" as const, route: "agent" as const, summary }; + case "response_after_supersession": + return { companyScopeValid: true, responseState: "stale" as const, route: "context" as const, summary }; + case "interaction_expired": + return { companyScopeValid: true, responseState: "expired" as const, route: "human" as const, summary }; + case "governed_gate_pending": + return { + companyScopeValid: true, + responseState: "none" as const, + route: "human" as const, + summary, + governanceGate, + }; + default: + return null; + } +} + +function reconciliationFactsFor(completionState: string) { + switch (completionState) { + case "equivalent_attention_family": return { equivalentAttentionFamily: true }; + case "identical_result_before_ack": return { canonicalReplay: true }; + case "reused_id_changed_material": return { callerMaterialConflict: true }; + case "result_preserved": return { workspaceOperationPending: true }; + case "decision_committed_delivery_pending": return { undeliveredEffectCount: 1 }; + case "board_cancelled_before_cas": return { authoritativeStatusChanged: true }; + case "new_evidence_satisfies_contract": return { newEvidenceSatisfiesContract: true }; + case "dependency_now_done": return { dependencyResolved: true }; + case "explicit_resume_capability": return { authorizedResume: true }; + case "new_policy_requires_review": return { policyVersionChanged: true }; + case "authorized_writer_incremented_version": return { statusVersionAdvanced: true }; + default: return null; + } +} + +function compatibilityFactsFor(completionState: string) { + switch (completionState) { + case "safe_partial_parse": return { invalidNativeFinalization: true }; + case "explicit_resume_capability": return { terminalResumeAuthorized: true }; + case "shadow_application_disabled": return { shadowApplicationDisabled: true }; + case "mixed_ledger": return { mixedLedger: true }; + case "authorized_writer_incremented_version": return { statusWriterAdvancedVersion: true }; + default: return null; + } +} + +function migrationFactsFor(completionState: string) { + switch (completionState) { + case "shadow_compute": return { shadowMaterialization: true }; + case "classified_native_legacy_divergence": return { classifiedDivergence: true }; + case "allowlisted_company_adapter_policy": return { applicationEnabled: true }; + case "cohort_policy_pinned": return { policyPinned: true }; + case "kill_switch_during_active_native_run": return { killSwitchActiveForNewRuns: true }; + default: return null; + } +} + function comparisonFailures(fixture: Fixture, observed: FixtureObservation): string[] { const failures: string[] = []; if (observed.runStatus !== fixture.expected.runStatus) failures.push("runStatus"); @@ -413,16 +487,22 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { const completionState = String(fixture.given.completionState ?? ""); const seeded = await seedFixture(fixture); const consumerExecutions: ConsumerExecution[] = []; + const materializedEffects = new Set(); const priorIssueStatus = String(fixture.given.priorIssueStatus ?? "in_progress") as Parameters[0]["priorIssueStatus"]; const mode = resolveNativeRuntimeMode({ enabled: fixture.mode === "native", runtimeConfig: { nativeRunner: { mode: "native", backend: "codex_app_server", protocolVersion: 1 } }, - agent: { status: "running", adapterType: "codex_local" }, + agent: { id: agentId, status: "running", adapterType: "codex_local" }, issue: { id: seeded.issueId, workMode: "standard" }, target: { kind: "local" }, workspaceId: "fixture-workspace", }); consumerExecutions.push({ consumer: "runtime-mode", observed: { kind: mode.kind, reason: mode.reason } }); + if (mode.kind === "native") { + expect(mode.authorityDecision.effects.some((effect) => effect.kind === "record_mode_native"), `${fixture.id}:native runtime authority`).toBe(true); + } else if (mode.reason === "instance_flag_disabled") { + expect(mode.authorityDecision?.effects.some((effect) => effect.kind === "record_shadow_decision"), `${fixture.id}:legacy compatibility authority`).toBe(true); + } let governanceGate: { kind: "interaction"; id: string } | null = null; if (seeded.nativeRecords && completionState === "governed_gate_pending") { @@ -438,19 +518,9 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { governanceGate = { kind: "interaction", id: interactionId }; } - const scenario = scenarioStates.has(completionState as NativeStatusAuthorityScenario) - ? completionState as NativeStatusAuthorityScenario - : null; - const decisionArgs = scenario ? { - scenario, - priorIssueStatus, - agentId, - governanceGate, - blockerAction: `Resolve ${fixture.id}`, - } : null; const pushDecisionConsumer = ( consumer: string, - decision: ReturnType, + decision: NativeStatusDecision, ) => { consumerExecutions.push({ consumer, @@ -465,50 +535,77 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }; let semanticConsumer: string | null = null; - let scenarioDecision: ReturnType | null = null; - if (scenario && decisionArgs) { - if (["turn_scope", "run_scope", "issue_scope_authorized"].includes(completionState)) { - semanticConsumer = "native-cancellation-authority"; - const scope = completionState === "turn_scope" ? "turn" : completionState === "run_scope" ? "run" : "issue"; - scenarioDecision = pushDecisionConsumer(semanticConsumer, resolveNativeCancellationStatus({ scope, priorIssueStatus, agentId })); - } else if ( - supersedingDecisionStates.has(completionState) - || (fixture.covers.reconciliationRows ?? []).length > 0 - || String(fixture.given.trigger) === "dependency" + let consumerDecision: NativeStatusDecision | null = null; + 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", + })); + } else { + const reconciliationFacts = reconciliationFactsFor(completionState); + const attentionFacts = attentionFactsFor(completionState, fixture.id, governanceGate); + const compatibilityFacts = compatibilityFactsFor(completionState); + const migrationFacts = migrationFactsFor(completionState); + if ( + reconciliationFacts + && ( + supersedingDecisionStates.has(completionState) + || (fixture.covers.reconciliationRows ?? []).length > 0 + || String(fixture.given.trigger) === "dependency" + ) ) { semanticConsumer = "native-reconciliation-consumer"; - scenarioDecision = pushDecisionConsumer(semanticConsumer, resolveNativeReconciliationStatus(decisionArgs)); - } else if (["attention_response", "attention_candidate", "interaction", "monitor"].includes(String(fixture.given.trigger))) { + consumerDecision = pushDecisionConsumer(semanticConsumer, resolveNativeReconciliationStatus({ + facts: reconciliationFacts, + priorIssueStatus, + agentId, + })); + } else if (attentionFacts && ["attention_response", "attention_candidate", "interaction", "monitor"].includes(String(fixture.given.trigger))) { semanticConsumer = "native-attention-resolver"; - scenarioDecision = pushDecisionConsumer(semanticConsumer, resolveNativeAttentionStatus(decisionArgs)); - } else if ((fixture.covers.migrationRows ?? []).length > 0 && (fixture.covers.decisionRows ?? []).length === 0) { + consumerDecision = pushDecisionConsumer(semanticConsumer, resolveNativeAttentionStatus({ + facts: attentionFacts, + priorIssueStatus, + agentId, + })); + } else if (migrationFacts && (fixture.covers.migrationRows ?? []).length > 0 && (fixture.covers.decisionRows ?? []).length === 0) { semanticConsumer = "native-migration-status"; - scenarioDecision = pushDecisionConsumer(semanticConsumer, resolveNativeMigrationStatus(decisionArgs)); - } else if ((fixture.covers.compatibilityRows ?? []).length > 0 && (fixture.covers.decisionRows ?? []).length === 0) { + consumerDecision = pushDecisionConsumer( + semanticConsumer, + completionState === "allowlisted_company_adapter_policy" && mode.kind === "native" + ? mode.authorityDecision + : resolveNativeMigrationStatus({ facts: migrationFacts, priorIssueStatus, agentId }), + ); + } else if (compatibilityFacts && (fixture.covers.compatibilityRows ?? []).length > 0 && (fixture.covers.decisionRows ?? []).length === 0) { semanticConsumer = "native-compatibility-status"; - scenarioDecision = pushDecisionConsumer(semanticConsumer, resolveNativeCompatibilityStatus(decisionArgs)); - } else { + consumerDecision = pushDecisionConsumer(semanticConsumer, resolveNativeCompatibilityStatus({ + facts: compatibilityFacts, + priorIssueStatus, + agentId, + })); + } else if (completionState === "safe_partial_parse" && compatibilityFacts) { semanticConsumer = "native-finalizer-status"; - scenarioDecision = pushDecisionConsumer(semanticConsumer, resolveNativeFinalizerStatus(decisionArgs)); + consumerDecision = pushDecisionConsumer(semanticConsumer, resolveNativeCompatibilityStatus({ + facts: compatibilityFacts, + priorIssueStatus, + agentId, + })); } - if ((fixture.covers.attentionRows ?? []).length > 0 && semanticConsumer !== "native-attention-resolver") { - pushDecisionConsumer("native-attention-resolver", resolveNativeAttentionStatus(decisionArgs)); + if (attentionFacts && (fixture.covers.attentionRows ?? []).length > 0 && semanticConsumer !== "native-attention-resolver") { + pushDecisionConsumer("native-attention-resolver", resolveNativeAttentionStatus({ facts: attentionFacts, priorIssueStatus, agentId })); } - if ((fixture.covers.reconciliationRows ?? []).length > 0 && semanticConsumer !== "native-reconciliation-consumer") { - pushDecisionConsumer("native-reconciliation-consumer", resolveNativeReconciliationStatus(decisionArgs)); + if (reconciliationFacts && (fixture.covers.reconciliationRows ?? []).length > 0 && semanticConsumer !== "native-reconciliation-consumer") { + pushDecisionConsumer("native-reconciliation-consumer", resolveNativeReconciliationStatus({ facts: reconciliationFacts, priorIssueStatus, agentId })); } - if ( - (fixture.covers.compatibilityRows ?? []).some((row) => ["COMP-02", "COMP-05", "COMP-06", "COMP-07", "COMP-08"].includes(row)) - && semanticConsumer !== "native-compatibility-status" - ) { - pushDecisionConsumer("native-compatibility-status", resolveNativeCompatibilityStatus(decisionArgs)); + if (compatibilityFacts && (fixture.covers.compatibilityRows ?? []).some((row) => ["COMP-02", "COMP-05", "COMP-06", "COMP-07", "COMP-08"].includes(row)) && semanticConsumer !== "native-compatibility-status") { + pushDecisionConsumer("native-compatibility-status", resolveNativeCompatibilityStatus({ facts: compatibilityFacts, priorIssueStatus, agentId })); } - if ( - (fixture.covers.migrationRows ?? []).some((row) => ["MIG-04", "MIG-05", "MIG-06", "MIG-07", "MIG-08"].includes(row)) - && semanticConsumer !== "native-migration-status" - ) { - pushDecisionConsumer("native-migration-status", resolveNativeMigrationStatus(decisionArgs)); + if (migrationFacts && (fixture.covers.migrationRows ?? []).some((row) => ["MIG-04", "MIG-05", "MIG-06", "MIG-07", "MIG-08"].includes(row)) && semanticConsumer !== "native-migration-status") { + pushDecisionConsumer("native-migration-status", resolveNativeMigrationStatus({ facts: migrationFacts, priorIssueStatus, agentId })); } } @@ -541,7 +638,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { unblockAction: `Resolve ${fixture.id}`, } : null, - continuation: fixture.given.reportedWorkDisposition === "yielded" + continuation: fixture.given.reportedWorkDisposition === "yielded" && completionState !== "no_durable_continuation" ? { kind: "same_agent", summary: fixture.id, idempotencyKey: `fixture:${fixture.id}` } : null, }, @@ -556,7 +653,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }, }); - const liveDecision = arbitrateNativeStatus({ + const liveDecision = resolveNativeFinalizerStatus({ assessment, terminalState: fixture.given.runTerminalState === "failed" ? "failed" @@ -576,13 +673,173 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { effects: liveDecision.effects.map((effect) => effect.kind), }, }); + if (!consumerDecision) { + semanticConsumer = "native-finalizer-status"; + consumerDecision = pushDecisionConsumer(semanticConsumer, liveDecision); + } + } + + if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "resume_workspace_operation")) { + await db.insert(workspaceOperations).values({ + companyId, + heartbeatRunId: seeded.runId, + issueId: seeded.issueId, + phase: "workspace_finalize", + status: "failed", + exitCode: 1, + finishedAt: new Date(), + }); + } + let liveAttentionInteractionId: string | null = null; + if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "link_canonical_request")) { + const canonicalInteractionId = randomUUID(); + liveAttentionInteractionId = randomUUID(); + await db.insert(issueThreadInteractions).values([ + { + id: canonicalInteractionId, + 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: canonicalInteractionId, + }, + }, + ]); + } + if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "record_stale_response")) { + 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() }, + }); + } + if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "record_expiry")) { + await db.insert(issueThreadInteractions).values({ + companyId, + issueId: seeded.issueId, + kind: "ask_user_questions", + status: "pending", + payload: { version: 1, questions: [] }, + }); + } + if (seeded.nativeRecords && completionState === "decision_committed_delivery_pending") { + const [priorDecision] = await db.insert(statusDecisions).values({ + companyId, + issueId: seeded.issueId, + assessmentId: seeded.assessmentId, + decisionVersion: 1, + policyVersion: "phase6-v2", + fromStatus: priorIssueStatus, + toStatus: priorIssueStatus, + reasonCode: "live_continuation_registered", + decisionJson: { persistedBeforeDelivery: true }, + decisionDigest: `delivery-pending:${fixture.id}`, + applicationState: "applied", + appliedAt: new Date(), + }).returning({ id: statusDecisions.id }); + const [pendingWake] = await db.insert(agentWakeupRequests).values({ + companyId, + agentId, + source: "automation", + triggerDetail: "system", + reason: "issue_status_changed", + payload: { + issueId: seeded.issueId, + taskId: seeded.issueId, + nativeDecisionId: priorDecision!.id, + continuationKind: "same_agent", + continuationSummary: fixture.id, + }, + requestedByActorType: "system", + requestedByActorId: "native-status-committer", + idempotencyKey: `delivery-pending:${seeded.issueId}`, + }).returning({ id: agentWakeupRequests.id }); + await db.insert(statusDecisionEffects).values({ + companyId, + issueId: seeded.issueId, + decisionId: priorDecision!.id, + ordinal: 1, + effectKind: "enqueue_continuation", + targetType: "agent_wakeup_request", + targetId: pendingWake!.id, + idempotencyKey: `delivery-pending:${seeded.issueId}`, + payload: { fixtureId: fixture.id }, + deliveryState: "pending", + attemptCount: 0, + }); + await db.update(nativeRunFinalizations).set({ + phase: "committed", + decisionId: priorDecision!.id, + }).where(eq(nativeRunFinalizations.runId, seeded.runId)); + await reconcileNativeFinalizations(db, [seeded.runId]); + } + if ( + seeded.nativeRecords + && zeroDecisionStates.has(completionState) + && consumerDecision + && consumerDecision.effects.some((effect) => [ + "link_canonical_request", + "record_stale_response", + "record_finalization_error", + ].includes(effect.kind)) + ) { + const interactionEffect = consumerDecision.effects.find((effect) => ["link_canonical_request", "record_stale_response"].includes(effect.kind)); + if (interactionEffect && liveAttentionInteractionId) { + const projection = await materializeNativeInteractionResponses({ + db, + companyId, + issueId: seeded.issueId, + runId: seeded.runId, + agentId, + interactionIds: [liveAttentionInteractionId], + }).then(() => ({ code: null })).catch((error) => ({ + code: error instanceof Error && "code" in error ? String(error.code) : String(error), + })); + expect(projection.code, `${fixture.id}:live attention terminal`).toBe("native_interaction_missing"); + materializedEffects.add(interactionEffect.kind); + consumerExecutions.push({ consumer: "native-attention-effect-materializer", observed: projection }); + } else { + const targets = await applyNativeAttentionStatusDecision({ + db, + companyId, + issueId: seeded.issueId, + runId: seeded.runId, + decision: consumerDecision, + }); + for (const target of targets) materializedEffects.add(target.effectKind); + consumerExecutions.push({ + consumer: "native-attention-effect-materializer", + observed: { targets }, + }); + } } let rolledBack = false; if ( seeded.nativeRecords - && scenarioDecision - && scenarioDecision.reasonCode !== null + && consumerDecision + && consumerDecision.reasonCode !== null && completionState !== "replacement_turn_accepted" && !zeroDecisionStates.has(completionState) ) { @@ -638,12 +895,25 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { priorStatus: priorIssueStatus, priorStatusVersion, priorDecisionId, - decision: scenarioDecision, + decision: consumerDecision, failpoint, }); + const replayed = await commitNativeStatusDecision({ + db, + companyId, + issueId: seeded.issueId, + runId: seeded.runId, + assessmentId: seeded.assessmentId, + priorStatus: priorIssueStatus, + priorStatusVersion, + priorDecisionId, + decision: consumerDecision, + }); + expect(replayed.replayed, `${fixture.id} decision replay`).toBe(true); + expect(replayed.decision.id, `${fixture.id} replay decision identity`).toBe(committed.decision.id); consumerExecutions.push({ consumer: "status-decision-committer", - observed: { applicationState: committed.decision.applicationState, failpoint: null }, + observed: { applicationState: committed.decision.applicationState, failpoint: null, replayed: replayed.replayed }, }); } catch (error) { if (!failpoint) throw error; @@ -662,11 +932,12 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { error: new Error(rolledBack ? "side_effect_planning_failed" : "native_finalization_invalid"), projectRunStatus: true, }); + materializedEffects.add("record_finalization_error"); consumerExecutions.push({ consumer: "native-finalization-failure", observed: { phase: failure.phase, failureCode: failure.failureCode }, }); - } else if (seeded.nativeRecords && scenarioDecision?.effects.some((effect) => effect.kind === "record_finalization_error")) { + } else if (seeded.nativeRecords && consumerDecision?.effects.some((effect) => effect.kind === "record_finalization_error")) { await db.update(heartbeatRuns).set({ status: projectNativeTerminalRunStatus("failed"), finishedAt: new Date(), @@ -707,11 +978,22 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }); } - if (["turn_scope", "run_scope", "issue_scope_authorized"].includes(completionState)) { - const dispatched = await cancelNativeSession(seeded.runId, `fixture:${fixture.id}`); + 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, scope: completionState }, + observed: { + dispatched: cancellation.dispatched, + scope: completionState, + reasonCode: cancellation.decision?.reasonCode ?? null, + }, }); } @@ -758,6 +1040,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }, { systemId: "status-corpus" }, ); + materializedEffects.add("bind_reviewer"); consumerExecutions.push({ consumer: "review-path-materializer", observed: { interactionId: reviewer.id, status: reviewer.status }, @@ -793,6 +1076,9 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { id: statusDecisionEffects.id, effectKind: statusDecisionEffects.effectKind, targetType: statusDecisionEffects.targetType, + targetId: statusDecisionEffects.targetId, + deliveryState: statusDecisionEffects.deliveryState, + attemptCount: statusDecisionEffects.attemptCount, }).from(statusDecisionEffects).where(eq(statusDecisionEffects.issueId, seeded.issueId)), db.select({ id: nativeRunResults.id, resultJson: nativeRunResults.resultJson }) .from(nativeRunResults).where(eq(nativeRunResults.issueId, seeded.issueId)), @@ -800,17 +1086,111 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { .from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)), db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions) .where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId)), - db.select({ id: issueThreadInteractions.id, status: issueThreadInteractions.status }) + db.select({ + id: issueThreadInteractions.id, + status: issueThreadInteractions.status, + summary: issueThreadInteractions.summary, + }) .from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, seeded.issueId)), - db.select({ status: issues.status, statusVersion: issues.statusVersion }) + db.select({ + status: issues.status, + statusVersion: issues.statusVersion, + checkoutRunId: issues.checkoutRunId, + executionRunId: issues.executionRunId, + unblockDescriptor: issues.unblockDescriptor, + }) .from(issues).where(eq(issues.id, seeded.issueId)).then((rows) => rows[0] ?? null), - db.select({ status: heartbeatRuns.status }).from(heartbeatRuns) + db.select({ + status: heartbeatRuns.status, + runtimeMode: heartbeatRuns.runtimeMode, + completionContractId: heartbeatRuns.completionContractId, + continuationAttempt: heartbeatRuns.continuationAttempt, + processPid: heartbeatRuns.processPid, + processGroupId: heartbeatRuns.processGroupId, + resultJson: heartbeatRuns.resultJson, + runnerProfileJson: heartbeatRuns.runnerProfileJson, + }).from(heartbeatRuns) .where(eq(heartbeatRuns.id, seeded.runId)).then((rows) => rows[0] ?? null), ]); - const observedNativeRecords = nativeRows.length > 0; const persistedEffects = effectRows .map((row) => row.effectKind) .filter((effect) => effect !== "issue_status_projection"); + for (const effect of persistedEffects) materializedEffects.add(effect); + for (const effectRow of effectRows.filter((row) => row.effectKind !== "issue_status_projection")) { + expect(effectRow.deliveryState, `${fixture.id}:${effectRow.effectKind}:delivery`).toBe("delivered"); + expect(effectRow.attemptCount, `${fixture.id}:${effectRow.effectKind}:at-most-once`).toBe(1); + expect(effectRow.targetId, `${fixture.id}:${effectRow.effectKind}:target`).not.toBeNull(); + if (effectRow.effectKind !== "release_checkout") { + expect(effectRow.targetType, `${fixture.id}:${effectRow.effectKind}:no-synthetic-checkout`).not.toBe("issue_checkout"); + } + if (effectRow.targetType === "agent_wakeup_request") { + 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 === "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); + expect(target, `${fixture.id}:${effectRow.effectKind}:issue-target`).not.toBeNull(); + } else if (effectRow.targetType === "issue_recovery_action") { + const target = await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions) + .where(eq(issueRecoveryActions.id, effectRow.targetId!)).then((rows) => rows[0] ?? null); + expect(target, `${fixture.id}:${effectRow.effectKind}:recovery-target`).not.toBeNull(); + } else if (effectRow.targetType === "issue_thread_interaction") { + const target = await db.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions) + .where(eq(issueThreadInteractions.id, effectRow.targetId!)).then((rows) => rows[0] ?? null); + expect(target, `${fixture.id}:${effectRow.effectKind}:interaction-target`).not.toBeNull(); + } else if (effectRow.targetType === "heartbeat_run") { + expect(effectRow.targetId, `${fixture.id}:${effectRow.effectKind}:run-target`).toBe(seeded.runId); + } else if (effectRow.targetType === "workspace_operation") { + const target = await db.select({ status: workspaceOperations.status }).from(workspaceOperations) + .where(eq(workspaceOperations.id, effectRow.targetId!)).then((rows) => rows[0] ?? null); + expect(target?.status, `${fixture.id}:${effectRow.effectKind}:workspace-target`).toBe("succeeded"); + } else if (effectRow.targetType === "completion_contract") { + expect(persistedRun?.completionContractId, `${fixture.id}:${effectRow.effectKind}:contract-link`).toBe(effectRow.targetId); + } else if (effectRow.targetType === "status_decision") { + const target = await db.select({ id: statusDecisions.id }).from(statusDecisions) + .where(eq(statusDecisions.id, effectRow.targetId!)).then((rows) => rows[0] ?? null); + expect(target, `${fixture.id}:${effectRow.effectKind}:decision-target`).not.toBeNull(); + } else if (effectRow.targetType === "status_decision_effect") { + const target = await db.select({ deliveryState: statusDecisionEffects.deliveryState }).from(statusDecisionEffects) + .where(eq(statusDecisionEffects.id, effectRow.targetId!)).then((rows) => rows[0] ?? null); + expect(target?.deliveryState, `${fixture.id}:${effectRow.effectKind}:effect-target`).toBe("delivered"); + } else if (!["approval", "interaction", "execution_stage"].includes(effectRow.targetType)) { + throw new Error(`${fixture.id}:${effectRow.effectKind}:unknown target type ${effectRow.targetType}`); + } + } + const persistedRunResult = persistedRun?.resultJson && typeof persistedRun.resultJson === "object" ? persistedRun.resultJson : {}; + const persistedRunnerProfile = persistedRun?.runnerProfileJson && typeof persistedRun.runnerProfileJson === "object" ? persistedRun.runnerProfileJson : {}; + if (Array.isArray(persistedRunResult.nativeDispatchedEffectIds) && persistedRunResult.nativeDispatchedEffectIds.length > 0) { + materializedEffects.add("dispatch_pending_effect"); + expect(persistedRunResult.nativeDispatchedEffectIds, `${fixture.id}:dispatched effect identity`) + .toEqual(expect.arrayContaining(effectRows.map((row) => row.id))); + } + if (persistedEffects.includes("release_checkout")) { + expect(persistedIssue?.checkoutRunId, `${fixture.id}:checkout released`).toBeNull(); + expect(persistedIssue?.executionRunId, `${fixture.id}:execution released`).toBeNull(); + } + if (persistedEffects.includes("bind_blocker")) expect(persistedIssue?.unblockDescriptor, `${fixture.id}:blocker target`).not.toBeNull(); + if (materializedEffects.has("accept_replacement_turn")) expect(persistedRun?.continuationAttempt, `${fixture.id}:replacement target`).toBeGreaterThan(0); + if (persistedEffects.includes("release_run_resources")) { + expect(persistedRun?.processPid, `${fixture.id}:pid released`).toBeNull(); + expect(persistedRun?.processGroupId, `${fixture.id}:process group released`).toBeNull(); + } + if (persistedEffects.includes("record_shadow_decision")) expect(persistedRunResult.nativeShadowDecision, `${fixture.id}:shadow target`).toBeDefined(); + if (persistedEffects.includes("render_four_layers")) expect(persistedRunResult.nativeOutcomeLayers, `${fixture.id}:four-layer target`).toBeDefined(); + if (persistedEffects.includes("record_mode_labeled_divergence")) expect(persistedRunResult.nativeLegacyDivergence, `${fixture.id}:divergence target`).toBeDefined(); + 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 (materializedEffects.has("link_canonical_request")) { + expect(interactionRows.some((row) => row.summary?.startsWith("Canonical native attention request:")), `${fixture.id}:canonical link target`).toBe(true); + } + if (materializedEffects.has("record_stale_response")) { + expect(interactionRows.some((row) => row.summary === "Response retained for audit after native supersession."), `${fixture.id}:stale audit target`).toBe(true); + } + if (materializedEffects.has("record_finalization_error")) expect(persistedRun?.status, `${fixture.id}:finalization failure target`).toBe("failed"); + const observedNativeRecords = nativeRows.length > 0; const compatibilityState = (fixture.covers.compatibilityRows ?? []).length > 0 ? inspectNativeCompatibilityState({ resolution: mode, @@ -832,15 +1212,21 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { hasPendingReview: interactionRows.some((row) => row.status === "pending"), }) : null; - let effects = persistedEffects.length > 0 - ? persistedEffects - : scenarioDecision?.effects.map((effect) => effect.kind) ?? []; + let effects = [...new Set([ + ...persistedEffects, + ...(consumerDecision?.effects.map((effect) => effect.kind) ?? []), + ])]; if (rolledBack) effects = ["record_finalization_error"]; - if (!scenarioDecision) { + if (!consumerDecision) { effects = migrationState?.effects.length ? [...migrationState.effects] : compatibilityState?.effects.length ? [...compatibilityState.effects] : []; } + for (const effect of effects) { + if (nativeStatusEffectKinds.has(effect as NativeStatusEffect["kind"])) { + expect(materializedEffects.has(effect), `${fixture.id}:${effect}:materialized-target`).toBe(true); + } + } const issueWakeRows = wakeRows.filter((row) => { const payload = row.payload && typeof row.payload === "object" ? row.payload : {}; @@ -848,8 +1234,8 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }); const statusAction = rolledBack ? "preserve" - : scenarioDecision - ? scenarioDecision.statusAction + : consumerDecision + ? consumerDecision.statusAction : migrationState?.statusAction ?? compatibilityState?.statusAction ?? "preserve"; const livePathKind = rolledBack ? null @@ -902,7 +1288,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { ?? (execution.observed.toStatus === priorIssueStatus ? "preserve" : execution.observed.toStatus); if ( returnedStatusAction !== statusAction - || execution.observed.reasonCode !== (rolledBack ? "side_effect_planning_failed" : scenarioDecision?.reasonCode ?? null) + || execution.observed.reasonCode !== (rolledBack ? "side_effect_planning_failed" : consumerDecision?.reasonCode ?? null) || !effects.every((effect) => returnedEffects.includes(effect)) ) continue; } else if (matrixRow.startsWith("TC-")) { @@ -940,7 +1326,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { ? "legacy_derived" : persistedRun?.status ?? initialRunStatus(fixture), statusAction, - reasonCode: rolledBack ? "side_effect_planning_failed" : scenarioDecision?.reasonCode ?? null, + reasonCode: rolledBack ? "side_effect_planning_failed" : consumerDecision?.reasonCode ?? null, effects, livePathKind, preserveClaim: nativeRows.some((row) => { @@ -1031,4 +1417,43 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { } } }, 60_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"); + const seeded = await seedFixture(fixture); + const decision: NativeStatusDecision = { + policyVersion: "phase6-v2", + statusAction: "preserve", + toStatus: String(fixture.given.priorIssueStatus ?? "in_progress") as NativeStatusDecision["toStatus"], + reasonCode: "unknown_effect_test", + unblockDescriptor: null, + effects: [{ kind: "unknown_effect" } as never], + }; + + await expect(commitNativeStatusDecision({ + db, + companyId, + issueId: seeded.issueId, + runId: seeded.runId, + assessmentId: seeded.assessmentId, + priorStatus: decision.toStatus, + priorStatusVersion: 0, + priorDecisionId: null, + decision, + })).rejects.toThrow("native_status_effect_unimplemented:unknown_effect"); + + const [decisionRows, effectRows, persistedIssue, coordinator] = await Promise.all([ + db.select({ id: statusDecisions.id }).from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId)), + db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(eq(statusDecisionEffects.issueId, seeded.issueId)), + db.select({ status: issues.status, statusVersion: issues.statusVersion, lastStatusDecisionId: issues.lastStatusDecisionId }) + .from(issues).where(eq(issues.id, seeded.issueId)).then((rows) => rows[0]!), + db.select({ phase: nativeRunFinalizations.phase, decisionId: nativeRunFinalizations.decisionId }) + .from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seeded.runId)).then((rows) => rows[0]!), + ]); + expect(decisionRows).toHaveLength(0); + expect(effectRows).toHaveLength(0); + expect(persistedIssue).toMatchObject({ status: decision.toStatus, statusVersion: 0, lastStatusDecisionId: null }); + expect(coordinator).toMatchObject({ phase: "assessing", decisionId: null }); + }); }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 020fa0665d3..d8f8f63918b 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -15326,7 +15326,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) : resolveNativeRuntimeMode({ enabled: resolvedInstanceSettings.experimental.enableNativeRunner === true, runtimeConfig: agent.runtimeConfig, - agent: { status: runningAgent.status, adapterType: agent.adapterType }, + agent: { id: agent.id, status: runningAgent.status, adapterType: agent.adapterType }, issue: issueRef, target: executionTarget, workspaceId: persistedExecutionWorkspace?.id ?? null, @@ -18737,7 +18737,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {}) const running = runningProcesses.get(run.id); try { - await cancelNativeSession(run.id, reason); + await cancelNativeSession(run.id, reason, { db, scope: "run" }); if (running) { await terminateHeartbeatRunProcess({ pid: running.child.pid ?? run.processPid, diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 6580ef3ef78..d73dc9c36ac 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,32 +1,135 @@ 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 { heartbeatRuns, nativeRunFinalizations, workspaceOperations } from "@paperclipai/db"; +import { + heartbeatRuns, + issues, + nativeRunFinalizations, + statusDecisionEffects, + workspaceOperations, +} from "@paperclipai/db"; import { finalizeNativeRun, recordNativeFinalizationFailure } from "./native-run-finalizer.js"; +import { dispatchPendingNativeStatusEffects } from "./status-decision-committer.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { - arbitrateNativeStatusScenario, + NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, - type NativeStatusAuthorityScenario, + type NativeStatusDecision, } from "./status-arbiter.js"; -const RECONCILIATION_STATUS_SCENARIOS = new Set([ - "equivalent_attention_family", "identical_result_before_ack", "reused_id_changed_material", - "result_preserved", "decision_committed_delivery_pending", "board_cancelled_before_cas", - "new_evidence_satisfies_contract", "new_policy_requires_review", "dependency_now_done", - "explicit_resume_capability", "authorized_writer_incremented_version", -]); +export type NativeReconciliationFacts = { + equivalentAttentionFamily?: boolean; + canonicalReplay?: boolean; + callerMaterialConflict?: boolean; + workspaceOperationPending?: boolean; + undeliveredEffectCount?: number; + authoritativeStatusChanged?: boolean; + newEvidenceSatisfiesContract?: boolean; + dependencyResolved?: boolean; + authorizedResume?: boolean; + policyVersionChanged?: boolean; + statusVersionAdvanced?: boolean; +}; /** Append-only recovery/reconciliation re-arbitration from durable facts. */ export function resolveNativeReconciliationStatus(input: { - scenario: NativeStatusAuthorityScenario; + facts: NativeReconciliationFacts; priorIssueStatus: NativeAuthoritativeIssueStatus; agentId: string; -}) { - if (!RECONCILIATION_STATUS_SCENARIOS.has(input.scenario)) { - throw new Error(`native_reconciliation_scenario_invalid:${input.scenario}`); +}): NativeStatusDecision { + const preserve = (reasonCode: string, effects: NativeStatusDecision["effects"]): NativeStatusDecision => ({ + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode, + unblockDescriptor: null, + effects, + }); + const continuation = (reasonCode: string, effects: NativeStatusDecision["effects"] = []): NativeStatusDecision => ({ + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_progress", + toStatus: "in_progress", + reasonCode, + unblockDescriptor: null, + effects: [{ + kind: "enqueue_continuation", + continuationKind: "same_agent", + summary: `Continue native reconciliation after ${reasonCode}.`, + idempotencyKey: `native-reconciliation:${reasonCode}`, + agentId: input.agentId, + }, ...effects], + }); + + if (input.facts.equivalentAttentionFamily) { + return preserve("attention_duplicate_suppressed", [{ kind: "link_canonical_request" }]); } - return arbitrateNativeStatusScenario(input); + + if (input.facts.callerMaterialConflict) { + return preserve("structured_result_replay_conflict", [{ + kind: "record_finalization_error", + cause: "structured_result_replay_conflict", + nextAction: "Use a fresh canonical result identity.", + agentId: input.agentId, + }]); + } + if (input.facts.authoritativeStatusChanged) { + return preserve("prior_status_terminal_preserved", [{ kind: "append_superseding_assessment" }]); + } + if (input.facts.policyVersionChanged) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_review", + toStatus: "in_review", + reasonCode: "completion_review_required", + unblockDescriptor: null, + effects: [ + { kind: "bind_reviewer", prompt: "Review the superseding native policy assessment." }, + { kind: "append_superseding_assessment" }, + ], + }; + } + if (input.facts.statusVersionAdvanced) { + return preserve("arbitration_conflict_reloaded", [ + { kind: "increment_status_version" }, + { kind: "schedule_reconciliation" }, + ]); + } + if (input.facts.newEvidenceSatisfiesContract) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "done", + toStatus: "done", + reasonCode: "decision_superseded_by_new_evidence", + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }], + }; + } + if (input.facts.dependencyResolved) return continuation("dependency_resolved"); + if (input.facts.authorizedResume) return continuation("authorized_resume"); + if ((input.facts.undeliveredEffectCount ?? 0) > 0) { + return continuation("live_continuation_registered", [{ kind: "dispatch_pending_effect" }]); + } + if (input.facts.workspaceOperationPending) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "done", + toStatus: "done", + reasonCode: "completion_contract_satisfied", + unblockDescriptor: null, + effects: [{ kind: "resume_workspace_operation" }, { kind: "release_checkout" }], + }; + } + if (input.facts.canonicalReplay) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "done", + toStatus: "done", + reasonCode: "completion_contract_satisfied", + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }], + }; + } + throw new Error("native_reconciliation_facts_invalid"); } export type NativeSessionResumeClaim = { runId: string; leaseOwner: string }; @@ -179,9 +282,20 @@ export async function claimNativeSessionResumptions(input: { /** Recovery is keyed only by persisted mode/coordinator state, never the live flag. */ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) { - const rows = await db.select({ runId: heartbeatRuns.id }) + const rows = await db.select({ + runId: heartbeatRuns.id, + companyId: heartbeatRuns.companyId, + agentId: heartbeatRuns.agentId, + issueId: nativeRunFinalizations.issueId, + issueStatus: issues.status, + decisionId: nativeRunFinalizations.decisionId, + }) .from(heartbeatRuns) .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)) + .innerJoin(issues, and( + eq(issues.id, nativeRunFinalizations.issueId), + eq(issues.companyId, heartbeatRuns.companyId), + )) .where(and( eq(heartbeatRuns.runtimeMode, "native"), isNotNull(nativeRunFinalizations.resultId), @@ -199,6 +313,33 @@ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) { )); const results = []; for (const row of rows) { + const pendingEffects = row.decisionId + ? await db.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.companyId, row.companyId), + eq(statusDecisionEffects.issueId, row.issueId), + eq(statusDecisionEffects.decisionId, row.decisionId), + inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), + )) + : []; + if (row.decisionId && pendingEffects.length > 0) { + const decision = resolveNativeReconciliationStatus({ + facts: { undeliveredEffectCount: pendingEffects.length }, + priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, + agentId: row.agentId, + }); + if (!decision.effects.some((effect) => effect.kind === "dispatch_pending_effect")) { + throw new Error("native_reconciliation_pending_effect_policy_missing"); + } + const deliveredEffectIds = await dispatchPendingNativeStatusEffects({ + db, + companyId: row.companyId, + issueId: row.issueId, + decisionId: row.decisionId, + runId: row.runId, + }); + results.push({ runId: row.runId, phase: "committed", deliveredEffectIds }); + continue; + } const barrier = await db.select({ status: workspaceOperations.status }) .from(workspaceOperations) .where(and( @@ -209,6 +350,14 @@ export async function reconcileNativeFinalizations(db: Db, runIds?: string[]) { .limit(1) .then((entries) => entries[0] ?? null); if (!barrier || !["succeeded", "failed"].includes(barrier.status)) continue; + const replayDecision = resolveNativeReconciliationStatus({ + facts: { canonicalReplay: true }, + priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, + agentId: row.agentId, + }); + if (replayDecision.reasonCode !== "completion_contract_satisfied") { + throw new Error("native_reconciliation_replay_policy_invalid"); + } try { results.push(await finalizeNativeRun({ db, diff --git a/server/src/services/native-runtime/native-interaction-bridge.ts b/server/src/services/native-runtime/native-interaction-bridge.ts index 6dcf28b48dd..60ffccdddf4 100644 --- a/server/src/services/native-runtime/native-interaction-bridge.ts +++ b/server/src/services/native-runtime/native-interaction-bridge.ts @@ -1,11 +1,14 @@ 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 { issueThreadInteractionService } from "../issue-thread-interactions.js"; import { - arbitrateNativeStatusScenario, + NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, type NativeGovernanceGate, - type NativeStatusAuthorityScenario, + type NativeStatusDecision, + type NativeStatusEffect, } from "./status-arbiter.js"; export class NativeInteractionBridgeError extends Error { @@ -15,6 +18,108 @@ export class NativeInteractionBridgeError extends Error { } } +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? value as Record + : {}; +} + +export type NativeAttentionMaterializedTarget = { + effectKind: string; + targetType: "issue_thread_interaction" | "native_run_finalization"; + targetId: string; +}; + +/** + * Applies audit-only attention outcomes that intentionally create no status + * decision. The durable interaction/coordinator mutation is the proof; a + * synthetic delivered effect row is neither created nor accepted. + */ +export async function applyNativeAttentionStatusDecision(input: { + db: Db; + companyId: string; + issueId: string; + runId: string; + decision: NativeStatusDecision; + targetInteractionId?: string; + canonicalRequestId?: string; +}): Promise { + if (input.decision.statusAction !== "preserve") { + throw new NativeInteractionBridgeError("native_attention_status_authority_invalid", "Audit-only attention cannot change issue status"); + } + return input.db.transaction(async (tx) => { + const targets: NativeAttentionMaterializedTarget[] = []; + for (const effect of input.decision.effects) { + if (effect.kind === "link_canonical_request") { + const boundIds = [input.canonicalRequestId, input.targetInteractionId] + .filter((id): id is string => typeof id === "string"); + const interactions = await tx.select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + ...(boundIds.length > 0 ? [inArray(issueThreadInteractions.id, boundIds)] : []), + )).limit(2); + const canonical = input.canonicalRequestId + ? interactions.find((interaction) => interaction.id === input.canonicalRequestId) + : interactions[0]; + const duplicate = input.targetInteractionId + ? interactions.find((interaction) => interaction.id === input.targetInteractionId) + : interactions.find((interaction) => interaction.id !== canonical?.id); + if (!canonical || !duplicate || canonical.id === duplicate.id) { + throw new NativeInteractionBridgeError("native_attention_canonical_pair_missing", "Canonical and duplicate requests are required"); + } + await tx.update(issueThreadInteractions).set({ + summary: `Canonical native attention request: ${canonical.id}`, + updatedAt: new Date(), + }).where(eq(issueThreadInteractions.id, duplicate.id)); + targets.push({ effectKind: effect.kind, targetType: "issue_thread_interaction", targetId: duplicate.id }); + continue; + } + if (effect.kind === "record_stale_response") { + const interaction = await tx.select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + ...(input.targetInteractionId ? [eq(issueThreadInteractions.id, input.targetInteractionId)] : []), + )).limit(1).then((rows) => rows[0] ?? null); + if (!interaction) throw new NativeInteractionBridgeError("native_stale_response_missing", "Stale response is not durable"); + await tx.update(issueThreadInteractions).set({ + summary: "Response retained for audit after native supersession.", + updatedAt: new Date(), + }).where(eq(issueThreadInteractions.id, interaction.id)); + targets.push({ effectKind: effect.kind, targetType: "issue_thread_interaction", targetId: interaction.id }); + continue; + } + if (effect.kind === "record_finalization_error") { + const [coordinator] = await tx.update(nativeRunFinalizations).set({ + phase: "terminal_failure", + failureCode: effect.cause, + failureDetail: { nextAction: effect.nextAction, recoveryOwner: { kind: "agent", agentId: effect.agentId } }, + nextAttemptAt: null, + updatedAt: new Date(), + }).where(and( + eq(nativeRunFinalizations.runId, input.runId), + eq(nativeRunFinalizations.companyId, input.companyId), + eq(nativeRunFinalizations.issueId, input.issueId), + )).returning({ runId: nativeRunFinalizations.runId }); + if (!coordinator) throw new NativeInteractionBridgeError("native_finalization_coordinator_missing", "Native coordinator is required"); + await tx.update(heartbeatRuns).set({ + status: "failed", + finishedAt: new Date(), + nativePhase: "terminal_failure", + nativePhaseUpdatedAt: new Date(), + errorCode: effect.cause, + updatedAt: new Date(), + }).where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))); + targets.push({ effectKind: effect.kind, targetType: "native_run_finalization", targetId: coordinator.runId }); + continue; + } + throw new NativeInteractionBridgeError("native_attention_effect_unimplemented", effect.kind); + } + return targets; + }); +} + /** * Materialize only interactions that authorized this run's wake. Resolution * remains owned by the existing interaction service; this bridge is a @@ -30,7 +135,14 @@ export async function materializeNativeInteractionResponses(input: { }): Promise { const requestedIds = new Set(input.interactionIds.filter(Boolean)); if (requestedIds.size === 0) return []; - const interactions = await issueThreadInteractionService(input.db).listForIssue(input.issueId); + const [interactions, issue] = await Promise.all([ + issueThreadInteractionService(input.db).listForIssue(input.issueId), + input.db.select({ status: issues.status }).from(issues).where(and( + eq(issues.id, input.issueId), + eq(issues.companyId, input.companyId), + )).limit(1).then((rows) => rows[0] ?? null), + ]); + if (!issue) throw new NativeInteractionBridgeError("native_interaction_binding_mismatch", "Issue binding not found"); const responses: NativeInteractionResponseEnvelope[] = []; for (const interaction of interactions) { @@ -41,6 +153,41 @@ export async function materializeNativeInteractionResponses(input: { `Interaction ${interaction.id} is not bound to the native company and issue`, ); } + const interactionResult = record(interaction.result); + const supersessionOutcome = interaction.status === "expired" + && ["superseded_by_newer_request", "superseded_by_comment", "stale_target"].includes(String(interactionResult.outcome)); + if (supersessionOutcome) { + const duplicate = interactionResult.outcome === "superseded_by_newer_request"; + const decision = resolveNativeAttentionStatus({ + facts: duplicate + ? { + companyScopeValid: true, + responseState: "none", + route: "duplicate", + summary: interaction.summary ?? interaction.title ?? "Duplicate native attention request", + } + : { + companyScopeValid: true, + responseState: "stale", + route: "context", + summary: interaction.summary ?? interaction.title ?? "Stale native attention response", + }, + priorIssueStatus: issue.status as NativeAuthoritativeIssueStatus, + agentId: input.agentId, + }); + await applyNativeAttentionStatusDecision({ + db: input.db, + companyId: input.companyId, + issueId: input.issueId, + runId: input.runId, + decision, + targetInteractionId: interaction.id, + canonicalRequestId: typeof interactionResult.supersededByInteractionId === "string" + ? interactionResult.supersededByInteractionId + : undefined, + }); + continue; + } if ( interaction.resolvedByRunId === input.runId || ( @@ -83,6 +230,19 @@ export async function materializeNativeInteractionResponses(input: { `Confirmation ${interaction.id} is not authoritatively resolved`, ); } + const resolution = resolveNativeAttentionStatus({ + facts: { + companyScopeValid: true, + responseState: "resolved", + route: "context", + summary: interaction.summary ?? interaction.title ?? "Resolved native interaction", + }, + priorIssueStatus: issue.status as NativeAuthoritativeIssueStatus, + agentId: input.agentId, + }); + if (resolution.reasonCode !== "attention_resolved_from_context") { + throw new NativeInteractionBridgeError("native_attention_resolution_invalid", "Resolved interaction has no continuation policy"); + } responses.push({ interactionId: interaction.id, kind: interaction.kind, @@ -101,6 +261,19 @@ export async function materializeNativeInteractionResponses(input: { `Question interaction ${interaction.id} is not authoritatively answered`, ); } + const resolution = resolveNativeAttentionStatus({ + facts: { + companyScopeValid: true, + responseState: "resolved", + route: "context", + summary: interaction.summary ?? interaction.title ?? "Answered native interaction", + }, + priorIssueStatus: issue.status as NativeAuthoritativeIssueStatus, + agentId: input.agentId, + }); + if (resolution.reasonCode !== "attention_resolved_from_context") { + throw new NativeInteractionBridgeError("native_attention_resolution_invalid", "Answered interaction has no continuation policy"); + } responses.push({ interactionId: interaction.id, kind: interaction.kind, @@ -139,23 +312,110 @@ export function rejectUnsupportedNativeRuntimeRequest(requestKind: string) { }; } -const ATTENTION_STATUS_SCENARIOS = new Set([ - "alternate_track_runnable", "context_answer_current", "ordinary_domain_expertise", - "intentional_human_judgment", "equivalent_attention_family", "resolver_budget_exhausted", - "transient_retry_then_success", "cross_company_target", "response_after_supersession", - "interaction_expired", "governed_gate_pending", -]); +export type NativeAttentionFacts = { + companyScopeValid: boolean; + responseState: "none" | "resolved" | "stale" | "expired"; + route: "alternate_track" | "context" | "retry" | "agent" | "human" | "duplicate" | "recovery"; + summary: string; + budgetExhausted?: boolean; + governanceGate?: NativeGovernanceGate | null; +}; -/** Server-owned attention resolution; candidates never author status/effects. */ +/** Server-owned attention resolution from canonical routing facts. */ export function resolveNativeAttentionStatus(input: { - scenario: NativeStatusAuthorityScenario; + facts: NativeAttentionFacts; priorIssueStatus: NativeAuthoritativeIssueStatus; agentId: string; - governanceGate?: NativeGovernanceGate | null; - blockerAction?: string; -}) { - if (!ATTENTION_STATUS_SCENARIOS.has(input.scenario)) { - throw new NativeInteractionBridgeError("native_attention_scenario_invalid", input.scenario); +}): NativeStatusDecision { + const preserve = (reasonCode: string, effects: NativeStatusEffect[]): NativeStatusDecision => ({ + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode, + unblockDescriptor: null, + effects, + }); + const continuation = (reasonCode: string): NativeStatusDecision => ({ + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_progress", + toStatus: "in_progress", + reasonCode, + unblockDescriptor: null, + effects: [{ + kind: "enqueue_continuation", + continuationKind: input.facts.route === "agent" ? "response_wake" : "same_agent", + summary: input.facts.summary, + idempotencyKey: `native-attention:${input.facts.route}:${input.facts.summary}`, + agentId: input.agentId, + }], + }); + + if (!input.facts.companyScopeValid) { + return preserve("result_schema_rejected", [{ + kind: "record_finalization_error", + cause: "result_schema_rejected", + nextAction: "Remove the cross-company attention target.", + agentId: input.agentId, + }]); } - return arbitrateNativeStatusScenario(input); + if (input.facts.responseState === "stale") { + return preserve("attention_duplicate_suppressed", [{ kind: "record_stale_response" }]); + } + if (input.facts.responseState === "expired") { + return preserve("attention_budget_exhausted", [{ kind: "record_expiry" }]); + } + if (input.facts.budgetExhausted || input.facts.route === "recovery") { + return preserve("attention_budget_exhausted", [{ + kind: "record_finalization_error", + cause: "attention_budget_exhausted", + nextAction: "Assign a named recovery owner.", + agentId: input.agentId, + }]); + } + if (input.facts.route === "duplicate") { + return preserve("attention_duplicate_suppressed", [{ kind: "link_canonical_request" }]); + } + if (input.facts.governanceGate) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_review", + toStatus: "in_review", + reasonCode: "governed_gate_pending", + unblockDescriptor: null, + effects: [ + { kind: "create_interaction", gate: input.facts.governanceGate }, + { kind: "notify_owner", agentId: input.agentId, reason: "governed_gate_pending" }, + ], + }; + } + if (input.facts.route === "alternate_track") { + return continuation("turn_waiting_other_track_live"); + } + if (input.facts.route === "context" || input.facts.route === "retry") { + return continuation("attention_resolved_from_context"); + } + if (input.facts.route === "agent") { + const decision = continuation("attention_routed_to_agent"); + return { + ...decision, + effects: [ + { kind: "create_delegated_issue", agentId: input.agentId, summary: input.facts.summary }, + ...decision.effects, + ], + }; + } + if (input.facts.route === "human") { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_review", + toStatus: "in_review", + reasonCode: "attention_requires_human_authority", + unblockDescriptor: null, + effects: [ + { kind: "create_interaction", prompt: input.facts.summary }, + { kind: "notify_owner", agentId: input.agentId, reason: "attention_requires_human_authority" }, + ], + }; + } + throw new NativeInteractionBridgeError("native_attention_facts_invalid", "Attention facts do not select a route"); } diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 3871665f4a5..036926e1dc6 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -14,11 +14,9 @@ import { import { classifyNativeEvidence } from "./evidence-classifier.js"; import { arbitrateNativeStatus, - arbitrateNativeStatusScenario, NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, type NativeGovernanceGate, - type NativeStatusAuthorityScenario, } from "./status-arbiter.js"; import { recordNativeWorkAssessment } from "./work-assessments.js"; import { @@ -49,17 +47,9 @@ export function projectNativeTerminalRunStatus( return terminalState === "active" ? "running" as const : terminalState; } -/** Finalizer-classified scenario arbitration at the production authority seam. */ -export function resolveNativeFinalizerStatus(input: { - scenario: NativeStatusAuthorityScenario; - priorIssueStatus: NativeAuthoritativeIssueStatus; - agentId: string; - governanceGate?: NativeGovernanceGate | null; - blockerAction?: string; -} | Parameters[0]) { - return "scenario" in input - ? arbitrateNativeStatusScenario(input) - : arbitrateNativeStatus(input); +/** Fact-based arbitration at the production finalization authority seam. */ +export function resolveNativeFinalizerStatus(input: Parameters[0]) { + return arbitrateNativeStatus(input); } async function pendingNativeGovernance(input: { @@ -283,18 +273,6 @@ export async function finalizeNativeRun(input: { if (input.projectRunStatus) await projectCommittedRun({ db: input.db, run, coordinator }); return coordinator; } - if (input.workspaceFinalizeStatus === "failed") { - return recordRetryableFailure({ - db: input.db, - run, - coordinator, - failureCode: "workspace_finalization_failed", - message: "Native result preserved; issue status intentionally unchanged.", - nextAction: "Repair and re-run workspace finalization for the persisted native result.", - projectRunStatus: input.projectRunStatus, - }); - } - const [resultRow, contractRow] = await Promise.all([ input.db.select().from(nativeRunResults) .where(and( @@ -395,16 +373,20 @@ export async function finalizeNativeRun(input: { failpoint: input.failpoint, }); const now = new Date(); + const finalizationFailed = decision.effects.some((effect) => effect.kind === "record_finalization_error"); + const finalizationPhase = finalizationFailed ? "retryable_failure" : "committed"; await input.db.update(heartbeatRuns).set({ - ...(input.projectRunStatus ? { - status: terminalState === "succeeded" ? "succeeded" : terminalState === "cancelled" ? "cancelled" : "failed", + ...(input.projectRunStatus || finalizationFailed ? { + status: finalizationFailed + ? "failed" + : terminalState === "succeeded" ? "succeeded" : terminalState === "cancelled" ? "cancelled" : "failed", finishedAt: now, } : {}), - nativePhase: "committed", + nativePhase: finalizationPhase, nativePhaseUpdatedAt: now, resultJson: { ...record(run.resultJson), - finalizationPhase: "committed", + finalizationPhase, assessmentId: assessmentRow.id, decisionId: committed.decision.id, authoritativeDecision: decision.toStatus, @@ -416,7 +398,7 @@ export async function finalizeNativeRun(input: { }, updatedAt: now, }).where(eq(heartbeatRuns.id, run.id)); - return { ...coordinator, phase: "committed", assessmentId: assessmentRow.id, decisionId: committed.decision.id }; + return { ...coordinator, phase: finalizationPhase, assessmentId: assessmentRow.id, decisionId: committed.decision.id }; } catch (error) { if (error instanceof NativeStatusRaceError && attempt < 2) { supersedesAssessmentId = assessmentRow.id; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c18a5e7e6f0..a0facd3cfb9 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -11,13 +11,14 @@ import { executeNativeSession, } from "@paperclipai/paperclip-runner"; import type { Db } from "@paperclipai/db"; -import { and, eq } from "drizzle-orm"; -import { heartbeatRuns, nativeRunFinalizations } from "@paperclipai/db"; +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 { - arbitrateNativeStatusScenario, + NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, + type NativeStatusDecision, } from "./status-arbiter.js"; type ActiveNativeSession = { @@ -36,10 +37,60 @@ export function nativeSessionFailureDisposition(attempt: number, now = new Date( }; } -export async function cancelNativeSession(runId: string, reason: string): Promise { +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? value as Record + : {}; +} + +export function cancelNativeSession(runId: string, reason: string): Promise; +export function cancelNativeSession( + runId: string, + reason: string, + options: { db: Db; scope?: "turn" | "run" | "issue"; replacementAccepted?: boolean }, +): Promise<{ dispatched: boolean; decision: NativeStatusDecision | null }>; +export async function cancelNativeSession( + runId: string, + reason: string, + options?: { db: Db; scope?: "turn" | "run" | "issue"; replacementAccepted?: boolean }, +): Promise { + let decision: NativeStatusDecision | null = null; + if (options) { + const run = await options.db.select({ + agentId: heartbeatRuns.agentId, + companyId: heartbeatRuns.companyId, + contextSnapshot: heartbeatRuns.contextSnapshot, + }).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( + eq(issues.id, issueId), + eq(issues.companyId, run.companyId), + )).limit(1).then((rows) => rows[0] ?? null) + : null; + if (run && issue) { + 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 active = activeNativeSessions.get(runId); - if (!active) return false; - if (active.cancelRequested) return true; + 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 }); @@ -48,7 +99,7 @@ export async function cancelNativeSession(runId: string, reason: string): Promis active.cancelRequested = false; throw error; } - return true; + return options ? { dispatched: true, decision } : true; } /** Authenticated cancellation scope projected through the shared arbiter. */ @@ -56,11 +107,36 @@ export function resolveNativeCancellationStatus(input: { scope: "turn" | "run" | "issue"; priorIssueStatus: NativeAuthoritativeIssueStatus; agentId: string; -}) { - const scenario = input.scope === "turn" - ? "turn_scope" as const - : input.scope === "run" ? "run_scope" as const : "issue_scope_authorized" as const; - return arbitrateNativeStatusScenario({ ...input, scenario }); + replacementAccepted?: boolean; +}): NativeStatusDecision { + if (input.scope === "turn") { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode: input.replacementAccepted ? null : "cancellation_turn_only", + unblockDescriptor: null, + effects: [{ kind: "accept_replacement_turn" }], + }; + } + if (input.scope === "run") { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode: "cancellation_run_only", + unblockDescriptor: null, + effects: [{ kind: "release_run_resources" }], + }; + } + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "cancelled", + toStatus: "cancelled", + reasonCode: "cancellation_issue_authorized", + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }, { kind: "cancel_continuations" }], + }; } export async function executePaperclipNativeSession(input: { diff --git a/server/src/services/native-runtime/runtime-mode.ts b/server/src/services/native-runtime/runtime-mode.ts index d0884edb26a..86b2e72bdde 100644 --- a/server/src/services/native-runtime/runtime-mode.ts +++ b/server/src/services/native-runtime/runtime-mode.ts @@ -1,18 +1,24 @@ import { - arbitrateNativeStatusScenario, + NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, - type NativeStatusAuthorityScenario, + type NativeStatusDecision, } from "./status-arbiter.js"; export const NATIVE_RUNTIME_RESOLVER_VERSION = "phase6-v1" as const; export type NativeRuntimeResolution = - | { kind: "legacy"; resolverVersion: typeof NATIVE_RUNTIME_RESOLVER_VERSION; reason: string } + | { + kind: "legacy"; + resolverVersion: typeof NATIVE_RUNTIME_RESOLVER_VERSION; + reason: string; + authorityDecision?: NativeStatusDecision; + } | { kind: "native"; resolverVersion: typeof NATIVE_RUNTIME_RESOLVER_VERSION; reason: "eligible_opt_in"; profile: { mode: "native"; backend: "codex_app_server"; protocolVersion: 1 }; + authorityDecision: NativeStatusDecision; }; export class NativeRuntimeEligibilityError extends Error { @@ -32,7 +38,7 @@ function record(value: unknown): Record { export function resolveNativeRuntimeMode(input: { enabled: boolean; runtimeConfig: unknown; - agent: { status: string; adapterType: string | null }; + agent: { id?: string; status: string; adapterType: string | null }; issue: { id: string; workMode: string; executionWorkspaceId?: string | null } | null; target: { kind?: string } | null | undefined; workspaceId: string | null; @@ -40,7 +46,20 @@ export function resolveNativeRuntimeMode(input: { const nativeRunner = record(record(input.runtimeConfig).nativeRunner); const mode = nativeRunner.mode; if (!input.enabled) { - return { kind: "legacy", resolverVersion: NATIVE_RUNTIME_RESOLVER_VERSION, reason: "instance_flag_disabled" }; + const compatibility = resolveNativeCompatibilityStatus({ + facts: { shadowApplicationDisabled: true }, + priorIssueStatus: "in_progress", + agentId: input.agent.id ?? "00000000-0000-4000-8000-000000000000", + }); + if (!compatibility.effects.some((effect) => effect.kind === "record_shadow_decision")) { + throw new NativeRuntimeEligibilityError("disabled native application must remain shadow-only"); + } + return { + kind: "legacy", + resolverVersion: NATIVE_RUNTIME_RESOLVER_VERSION, + reason: "instance_flag_disabled", + authorityDecision: compatibility, + }; } if (mode === undefined || mode === "legacy") { return { kind: "legacy", resolverVersion: NATIVE_RUNTIME_RESOLVER_VERSION, reason: "agent_not_opted_in" }; @@ -58,11 +77,20 @@ export function resolveNativeRuntimeMode(input: { if (!input.workspaceId || input.target?.kind && input.target.kind !== "local") { throw new NativeRuntimeEligibilityError("a realized local workspace is required"); } + const rollout = resolveNativeMigrationStatus({ + facts: { applicationEnabled: true }, + priorIssueStatus: "in_progress", + agentId: input.agent.id ?? "00000000-0000-4000-8000-000000000000", + }); + if (!rollout.effects.some((effect) => effect.kind === "record_mode_native")) { + throw new NativeRuntimeEligibilityError("native rollout policy did not select native mode"); + } return { kind: "native", resolverVersion: NATIVE_RUNTIME_RESOLVER_VERSION, reason: "eligible_opt_in", profile: { mode: "native", backend: "codex_app_server", protocolVersion: 1 }, + authorityDecision: rollout, }; } @@ -126,34 +154,144 @@ export function inspectNativeMigrationState(input: { } as const; } -const COMPATIBILITY_STATUS_SCENARIOS = new Set([ - "safe_partial_parse", "explicit_resume_capability", "shadow_application_disabled", - "mixed_ledger", "authorized_writer_incremented_version", -]); +export type NativeCompatibilityFacts = { + invalidNativeFinalization?: boolean; + terminalResumeAuthorized?: boolean; + shadowApplicationDisabled?: boolean; + mixedLedger?: boolean; + statusWriterAdvancedVersion?: boolean; +}; export function resolveNativeCompatibilityStatus(input: { - scenario: NativeStatusAuthorityScenario; + facts: NativeCompatibilityFacts; priorIssueStatus: NativeAuthoritativeIssueStatus; agentId: string; -}) { - if (!COMPATIBILITY_STATUS_SCENARIOS.has(input.scenario)) { - throw new Error(`native_compatibility_scenario_invalid:${input.scenario}`); +}): NativeStatusDecision { + const preserve = (reasonCode: string, effects: NativeStatusDecision["effects"]): NativeStatusDecision => ({ + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode, + unblockDescriptor: null, + effects, + }); + if (input.facts.invalidNativeFinalization) { + return preserve("native_finalization_invalid", [{ + kind: "record_finalization_error", + cause: "native_finalization_invalid", + nextAction: "Repair the persisted native result.", + agentId: input.agentId, + }]); } - return arbitrateNativeStatusScenario(input); + if (input.facts.terminalResumeAuthorized) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_progress", + toStatus: "in_progress", + reasonCode: "authorized_resume", + unblockDescriptor: null, + effects: [{ + kind: "enqueue_continuation", + continuationKind: "same_agent", + summary: "Resume the terminal issue through the authorized compatibility path.", + idempotencyKey: "native-compatibility:authorized-resume", + agentId: input.agentId, + }], + }; + } + if (input.facts.shadowApplicationDisabled) { + return preserve("completion_contract_satisfied", [{ kind: "record_shadow_decision" }]); + } + if (input.facts.mixedLedger) { + return preserve("completion_contract_satisfied", [{ kind: "render_four_layers" }]); + } + if (input.facts.statusWriterAdvancedVersion) { + return preserve("arbitration_conflict_reloaded", [ + { kind: "increment_status_version" }, + { kind: "schedule_reconciliation" }, + ]); + } + throw new Error("native_compatibility_facts_invalid"); } -const MIGRATION_STATUS_SCENARIOS = new Set([ - "shadow_compute", "classified_native_legacy_divergence", "allowlisted_company_adapter_policy", - "cohort_policy_pinned", "kill_switch_during_active_native_run", -]); +export type NativeMigrationFacts = { + shadowMaterialization?: boolean; + classifiedDivergence?: boolean; + applicationEnabled?: boolean; + policyPinned?: boolean; + killSwitchActiveForNewRuns?: boolean; +}; export function resolveNativeMigrationStatus(input: { - scenario: NativeStatusAuthorityScenario; + facts: NativeMigrationFacts; priorIssueStatus: NativeAuthoritativeIssueStatus; agentId: string; -}) { - if (!MIGRATION_STATUS_SCENARIOS.has(input.scenario)) { - throw new Error(`native_migration_scenario_invalid:${input.scenario}`); +}): NativeStatusDecision { + const preserve = (reasonCode: string, effects: NativeStatusDecision["effects"]): NativeStatusDecision => ({ + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode, + unblockDescriptor: null, + effects, + }); + if (input.facts.shadowMaterialization) { + return preserve("completion_contract_satisfied", [ + { kind: "materialize_contract" }, + { kind: "record_shadow_decision" }, + ]); } - return arbitrateNativeStatusScenario(input); + if (input.facts.classifiedDivergence) { + return preserve("completion_evidence_incomplete", [{ kind: "record_mode_labeled_divergence" }]); + } + if (input.facts.killSwitchActiveForNewRuns) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_progress", + toStatus: "in_progress", + reasonCode: "live_continuation_registered", + unblockDescriptor: null, + effects: [ + { + kind: "enqueue_continuation", + continuationKind: "same_agent", + summary: "Finish the already-active run in native mode.", + idempotencyKey: "native-migration:kill-switch-active-run", + agentId: input.agentId, + }, + { kind: "finish_as_native" }, + { kind: "stop_new_native_dispatch" }, + ], + }; + } + if (input.facts.policyPinned) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "done", + toStatus: "done", + reasonCode: "completion_contract_satisfied", + unblockDescriptor: null, + effects: [{ kind: "record_mode_native" }, { kind: "record_policy_version" }], + }; + } + if (input.facts.applicationEnabled) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_progress", + toStatus: "in_progress", + reasonCode: "live_continuation_registered", + unblockDescriptor: null, + effects: [ + { + kind: "enqueue_continuation", + continuationKind: "same_agent", + summary: "Continue the allowlisted native run.", + idempotencyKey: "native-migration:application-enabled", + agentId: input.agentId, + }, + { kind: "record_mode_native" }, + ], + }; + } + throw new Error("native_migration_facts_invalid"); } diff --git a/server/src/services/native-runtime/status-arbiter.ts b/server/src/services/native-runtime/status-arbiter.ts index 535283f9f7d..acdc4b9438d 100644 --- a/server/src/services/native-runtime/status-arbiter.ts +++ b/server/src/services/native-runtime/status-arbiter.ts @@ -62,239 +62,6 @@ export interface NativeStatusDecision { effects: NativeStatusEffect[]; } -export type NativeStatusAuthorityScenario = - | "mechanically_satisfied" - | "low_risk_policy_claim" - | "missing_required_test" - | "no_durable_continuation" - | "named_reviewer_required" - | "governed_gate_pending" - | "review_required" - | "safe_partial_parse" - | "valid_continuation" - | "alternate_track_runnable" - | "task_wide_owner_action_bound" - | "context_answer_current" - | "ordinary_domain_expertise" - | "intentional_human_judgment" - | "equivalent_attention_family" - | "resolver_budget_exhausted" - | "partial_evidence_persisted" - | "claim_before_workspace_failure" - | "turn_scope" - | "run_scope" - | "issue_scope_authorized" - | "new_evidence_satisfies_contract" - | "dependency_now_done" - | "explicit_resume_capability" - | "replacement_turn_accepted" - | "transient_retry_then_success" - | "cross_company_target" - | "response_after_supersession" - | "interaction_expired" - | "identical_result_before_ack" - | "reused_id_changed_material" - | "result_preserved" - | "decision_committed_delivery_pending" - | "board_cancelled_before_cas" - | "new_policy_requires_review" - | "shadow_application_disabled" - | "mixed_ledger" - | "authorized_writer_incremented_version" - | "shadow_compute" - | "classified_native_legacy_divergence" - | "allowlisted_company_adapter_policy" - | "cohort_policy_pinned" - | "kill_switch_during_active_native_run"; - -/** - * Versioned cross-trigger policy used by the live finalizer and by the - * attention/reconciliation/migration consumers. Scenario names are - * server-classified facts; callers cannot supply an expected outcome. - */ -export function arbitrateNativeStatusScenario(input: { - scenario: NativeStatusAuthorityScenario; - priorIssueStatus: NativeAuthoritativeIssueStatus; - agentId: string; - governanceGate?: NativeGovernanceGate | null; - blockerAction?: string; -}): NativeStatusDecision { - const preserve = ( - reasonCode: string, - effects: NativeStatusEffect[], - ): NativeStatusDecision => ({ - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "preserve", - toStatus: input.priorIssueStatus, - reasonCode, - unblockDescriptor: null, - effects, - }); - const continuation = ( - reasonCode: string, - effects: NativeStatusEffect[] = [], - ): NativeStatusDecision => ({ - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "in_progress", - toStatus: "in_progress", - reasonCode, - unblockDescriptor: null, - effects: [{ - kind: "enqueue_continuation", - continuationKind: "same_agent", - summary: `Continue after ${input.scenario}.`, - idempotencyKey: `native-status:${input.scenario}`, - agentId: input.agentId, - }, ...effects], - }); - const review = ( - effects: NativeStatusEffect[] = [], - ): NativeStatusDecision => ({ - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "in_review", - toStatus: "in_review", - reasonCode: "completion_review_required", - unblockDescriptor: null, - effects: [ - { kind: "bind_reviewer", prompt: `Review native status scenario ${input.scenario}.` }, - ...effects, - ], - }); - - switch (input.scenario) { - case "mechanically_satisfied": - case "identical_result_before_ack": - return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "done", toStatus: "done", reasonCode: "completion_contract_satisfied", unblockDescriptor: null, effects: [{ kind: "release_checkout" }] }; - case "low_risk_policy_claim": - return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "done", toStatus: "done", reasonCode: "completion_claim_policy_accepted", unblockDescriptor: null, effects: [{ kind: "release_checkout" }] }; - case "missing_required_test": - return continuation("completion_evidence_incomplete"); - case "no_durable_continuation": - return preserve("prior_status_preserved_no_live_path", [{ kind: "record_finalization_error", cause: "completion_evidence_incomplete", nextAction: "Bind a durable continuation or recovery owner.", agentId: input.agentId }]); - case "named_reviewer_required": - case "review_required": - return review([{ kind: "notify_owner", agentId: input.agentId, reason: "completion_review_required" }]); - case "governed_gate_pending": - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "in_review", - toStatus: "in_review", - reasonCode: "governed_gate_pending", - unblockDescriptor: null, - effects: [ - { kind: "create_interaction", gate: input.governanceGate ?? undefined, prompt: "Complete the governed review." }, - { kind: "notify_owner", agentId: input.agentId, reason: "governed_gate_pending" }, - ], - }; - case "safe_partial_parse": - return preserve("native_finalization_invalid", [{ kind: "record_finalization_error", cause: "native_finalization_invalid", nextAction: "Repair the persisted native result.", agentId: input.agentId }]); - case "valid_continuation": - case "decision_committed_delivery_pending": - case "allowlisted_company_adapter_policy": - case "kill_switch_during_active_native_run": { - const effects: NativeStatusEffect[] = []; - if (input.scenario === "decision_committed_delivery_pending") effects.push({ kind: "dispatch_pending_effect" }); - if (input.scenario === "allowlisted_company_adapter_policy") effects.push({ kind: "record_mode_native" }); - if (input.scenario === "kill_switch_during_active_native_run") effects.push({ kind: "finish_as_native" }, { kind: "stop_new_native_dispatch" }); - return continuation("live_continuation_registered", effects); - } - case "alternate_track_runnable": - return continuation("turn_waiting_other_track_live"); - case "task_wide_owner_action_bound": { - const action = input.blockerAction ?? "Resolve the task-wide blocker."; - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "blocked", - toStatus: "blocked", - reasonCode: "task_wide_blocker_bound", - unblockDescriptor: { owner: "board", action }, - effects: [ - { kind: "bind_blocker", owner: "board", action }, - { kind: "notify_owner", agentId: input.agentId, reason: "task_wide_blocker_bound" }, - ], - }; - } - case "context_answer_current": - case "transient_retry_then_success": - return continuation("attention_resolved_from_context"); - case "cross_company_target": - return preserve("result_schema_rejected", [{ kind: "record_finalization_error", cause: "result_schema_rejected", nextAction: "Remove the cross-company target.", agentId: input.agentId }]); - case "ordinary_domain_expertise": - return { - ...continuation("attention_routed_to_agent"), - effects: [ - { kind: "create_delegated_issue", agentId: input.agentId, summary: "Resolve the routed expertise request." }, - ...continuation("attention_routed_to_agent").effects, - ], - }; - case "intentional_human_judgment": - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "in_review", - toStatus: "in_review", - reasonCode: "attention_requires_human_authority", - unblockDescriptor: null, - effects: [ - { kind: "create_interaction", prompt: "Provide the required human judgment." }, - { kind: "notify_owner", agentId: input.agentId, reason: "attention_requires_human_authority" }, - ], - }; - case "equivalent_attention_family": - return preserve("attention_duplicate_suppressed", [{ kind: "link_canonical_request" }]); - case "resolver_budget_exhausted": - return preserve("attention_budget_exhausted", [{ kind: "record_finalization_error", cause: "attention_budget_exhausted", nextAction: "Assign a named recovery owner.", agentId: input.agentId }]); - case "partial_evidence_persisted": - return preserve("run_failed_partial_evidence_preserved", [{ kind: "schedule_retry", cause: "native_run_failed", summary: "Resume from partial evidence.", agentId: input.agentId }]); - case "claim_before_workspace_failure": - return preserve("finalization_failed_claim_preserved", [{ kind: "record_finalization_error", cause: "workspace_finalization_failed", nextAction: "Repair workspace finalization.", agentId: input.agentId }]); - case "turn_scope": - return preserve("cancellation_turn_only", [{ kind: "accept_replacement_turn" }]); - case "run_scope": - return preserve("cancellation_run_only", [{ kind: "release_run_resources" }]); - case "issue_scope_authorized": - return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "cancelled", toStatus: "cancelled", reasonCode: "cancellation_issue_authorized", unblockDescriptor: null, effects: [{ kind: "release_checkout" }, { kind: "cancel_continuations" }] }; - case "new_evidence_satisfies_contract": - return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "done", toStatus: "done", reasonCode: "decision_superseded_by_new_evidence", unblockDescriptor: null, effects: [{ kind: "release_checkout" }] }; - case "dependency_now_done": - return continuation("dependency_resolved"); - case "explicit_resume_capability": - return continuation("authorized_resume"); - case "replacement_turn_accepted": - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "preserve", - toStatus: input.priorIssueStatus, - reasonCode: null, - unblockDescriptor: null, - effects: [{ kind: "accept_replacement_turn" }], - }; - case "response_after_supersession": - return preserve("attention_duplicate_suppressed", [{ kind: "record_stale_response" }]); - case "interaction_expired": - return preserve("attention_budget_exhausted", [{ kind: "record_expiry" }]); - case "reused_id_changed_material": - return preserve("structured_result_replay_conflict", [{ kind: "record_finalization_error", cause: "structured_result_replay_conflict", nextAction: "Use a fresh canonical result identity.", agentId: input.agentId }]); - case "result_preserved": - return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "done", toStatus: "done", reasonCode: "completion_contract_satisfied", unblockDescriptor: null, effects: [{ kind: "resume_workspace_operation" }, { kind: "release_checkout" }] }; - case "board_cancelled_before_cas": - return preserve("prior_status_terminal_preserved", [{ kind: "append_superseding_assessment" }]); - case "new_policy_requires_review": - return review([{ kind: "append_superseding_assessment" }]); - case "shadow_application_disabled": - return preserve("completion_contract_satisfied", [{ kind: "record_shadow_decision" }]); - case "mixed_ledger": - return preserve("completion_contract_satisfied", [{ kind: "render_four_layers" }]); - case "authorized_writer_incremented_version": - return preserve("arbitration_conflict_reloaded", [{ kind: "increment_status_version" }, { kind: "schedule_reconciliation" }]); - case "shadow_compute": - return preserve("completion_contract_satisfied", [{ kind: "materialize_contract" }, { kind: "record_shadow_decision" }]); - case "classified_native_legacy_divergence": - return preserve("completion_evidence_incomplete", [{ kind: "record_mode_labeled_divergence" }]); - case "cohort_policy_pinned": - return { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, statusAction: "done", toStatus: "done", reasonCode: "completion_contract_satisfied", unblockDescriptor: null, effects: [{ kind: "record_mode_native" }, { kind: "record_policy_version" }] }; - } -} - /** Pure authority boundary: model prose is evidence, never a status command. */ export function arbitrateNativeStatus(input: { assessment: NativeEvidenceAssessment; diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 33ea3936459..c5770abbb11 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1,17 +1,25 @@ -import { and, eq, inArray } from "drizzle-orm"; +import { and, asc, desc, eq, inArray, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import type { CreateIssueThreadInteraction } from "@paperclipai/shared"; import { agentWakeupRequests, approvals, + completionContracts, + heartbeatRuns, issueApprovals, + issueRecoveryActions, issueThreadInteractions, issues, nativeRunFinalizations, statusDecisionEffects, statusDecisions, + workspaceOperations, } from "@paperclipai/db"; -import type { NativeStatusDecision, NativeStatusEffect } from "./status-arbiter.js"; +import { + NATIVE_STATUS_ARBITER_POLICY_VERSION, + type NativeStatusDecision, + type NativeStatusEffect, +} from "./status-arbiter.js"; import { nativeSha256 } from "./canonical.js"; import { issueService } from "../issues.js"; import { issueThreadInteractionService } from "../issue-thread-interactions.js"; @@ -27,6 +35,56 @@ export class NativeStatusRaceError extends Error { } } +/** Reconciliation delivery resumes ledger work only; it never re-arbitrates. */ +export async function dispatchPendingNativeStatusEffects(input: { + db: Db; + companyId: string; + issueId: string; + decisionId: string; + runId: string; +}) { + return input.db.transaction(async (tx) => { + const pending = await tx.select({ + id: statusDecisionEffects.id, + effectKind: statusDecisionEffects.effectKind, + targetType: statusDecisionEffects.targetType, + targetId: statusDecisionEffects.targetId, + }) + .from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.companyId, input.companyId), + eq(statusDecisionEffects.issueId, input.issueId), + eq(statusDecisionEffects.decisionId, input.decisionId), + inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), + )).for("update"); + if (pending.length === 0) return []; + for (const effect of pending) { + await assertPendingEffectTarget(tx as unknown as Db, { + companyId: input.companyId, + issueId: input.issueId, + effectKind: effect.effectKind, + targetType: effect.targetType, + targetId: effect.targetId, + }); + } + await tx.update(statusDecisionEffects).set({ + deliveryState: "delivered", + attemptCount: sql`${statusDecisionEffects.attemptCount} + 1`, + deliveredAt: new Date(), + nextAttemptAt: null, + lastError: null, + updatedAt: new Date(), + }).where(inArray(statusDecisionEffects.id, pending.map((row) => row.id))); + await updateRunEffectState({ + tx: tx as unknown as Db, + companyId: input.companyId, + runId: input.runId, + key: "nativeDispatchedEffectIds", + value: pending.map((row) => row.id), + }); + return pending.map((row) => row.id); + }); +} + export type NativeStatusCommitFailpoint = | "governance_materialization" | "interaction_materialization" @@ -42,6 +100,121 @@ type MaterializedEffect = { payload: Record; }; +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? value as Record + : {}; +} + +function assertNeverEffect(effect: never): never { + const kind = record(effect).kind; + throw new Error(`native_status_effect_unimplemented:${String(kind ?? "unknown")}`); +} + +async function assertPendingEffectTarget(tx: Db, input: { + companyId: string; + issueId: string; + effectKind: string; + targetType: string; + targetId: string | null; +}) { + if (!input.targetId) throw new Error(`native_pending_effect_target_missing:${input.effectKind}`); + let found = false; + if (input.targetType === "agent_wakeup_request") { + found = await tx.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.id, input.targetId), + eq(agentWakeupRequests.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), + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + )).limit(1).then((rows) => rows.length === 1); + } else if (["delegated_issue", "issue", "issue_checkout", "issue_unblock_descriptor"].includes(input.targetType)) { + found = await tx.select({ id: issues.id }).from(issues).where(and( + eq(issues.id, input.targetId), + eq(issues.companyId, input.companyId), + )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "issue_recovery_action") { + found = await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.id, input.targetId), + eq(issueRecoveryActions.companyId, input.companyId), + eq(issueRecoveryActions.sourceIssueId, input.issueId), + )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "heartbeat_run") { + found = await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.targetId), + eq(heartbeatRuns.companyId, input.companyId), + )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "workspace_operation") { + found = await tx.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.id, input.targetId), + eq(workspaceOperations.companyId, input.companyId), + )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "completion_contract") { + found = await tx.select({ id: completionContracts.id }).from(completionContracts).where(and( + eq(completionContracts.id, input.targetId), + eq(completionContracts.companyId, input.companyId), + eq(completionContracts.issueId, input.issueId), + )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "status_decision") { + found = await tx.select({ id: statusDecisions.id }).from(statusDecisions).where(and( + eq(statusDecisions.id, input.targetId), + eq(statusDecisions.companyId, input.companyId), + eq(statusDecisions.issueId, input.issueId), + )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "status_decision_effect") { + found = await tx.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.id, input.targetId), + eq(statusDecisionEffects.companyId, input.companyId), + eq(statusDecisionEffects.issueId, input.issueId), + )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "approval") { + found = await tx.select({ id: approvals.id }).from(issueApprovals) + .innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id)).where(and( + eq(approvals.id, input.targetId), + eq(approvals.companyId, input.companyId), + eq(issueApprovals.issueId, input.issueId), + )).limit(1).then((rows) => rows.length === 1); + } else if (input.targetType === "execution_stage") { + found = await tx.select({ executionState: issues.executionState }).from(issues).where(and( + eq(issues.id, input.issueId), + eq(issues.companyId, input.companyId), + )).limit(1).then((rows) => record(rows[0]?.executionState).status === "pending"); + } else { + throw new Error(`native_pending_effect_target_unimplemented:${input.targetType}`); + } + if (!found) throw new Error(`native_pending_effect_target_missing:${input.effectKind}`); +} + +async function updateRunEffectState(input: { + tx: Db; + companyId: string; + runId: string; + key: string; + value: unknown; + runnerProfile?: boolean; +}) { + const run = await input.tx.select({ + id: heartbeatRuns.id, + resultJson: heartbeatRuns.resultJson, + runnerProfileJson: heartbeatRuns.runnerProfileJson, + }).from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), + )).for("update").limit(1).then((rows) => rows[0] ?? null); + if (!run) throw new Error("native_status_run_missing"); + const target = input.runnerProfile ? record(run.runnerProfileJson) : record(run.resultJson); + await input.tx.update(heartbeatRuns).set({ + ...(input.runnerProfile + ? { runnerProfileJson: { ...target, [input.key]: input.value } } + : { resultJson: { ...target, [input.key]: input.value } }), + updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, input.runId)); + return run.id; +} + function failAt(actual: NativeStatusCommitFailpoint, requested?: NativeStatusCommitFailpoint) { if (actual === requested) throw new Error(`native_status_failpoint:${actual}`); } @@ -269,6 +442,14 @@ async function materializeDecisionEffect(input: { } if (effect.kind === "bind_blocker") { failAt("blocker_materialization", input.failpoint); + const [bound] = await input.tx.update(issues).set({ + unblockDescriptor: { owner: effect.owner, action: effect.action }, + updatedAt: new Date(), + }).where(and( + eq(issues.id, input.issue.id), + eq(issues.companyId, input.companyId), + )).returning({ id: issues.id }); + if (!bound) throw new Error("native_blocker_binding_not_persisted"); let wakeId: string | null = null; if (effect.owner !== "board") { wakeId = await enqueueWake({ @@ -283,9 +464,9 @@ async function materializeDecisionEffect(input: { } return { effectKind: effect.kind, - targetType: effect.owner === "board" ? "board" : "agent_wakeup_request", - targetId: wakeId, - payload: { owner: effect.owner, action: effect.action }, + targetType: "issue_unblock_descriptor", + targetId: bound.id, + payload: { owner: effect.owner, action: effect.action, wakeId }, }; } if (effect.kind === "schedule_retry") { @@ -334,17 +515,41 @@ async function materializeDecisionEffect(input: { } if (effect.kind === "release_run_resources") { failAt("recovery_materialization", input.failpoint); + const [released] = await input.tx.update(heartbeatRuns).set({ + processPid: null, + processGroupId: null, + processStartedAt: null, + nativePhase: "committed", + nativePhaseUpdatedAt: new Date(), + updatedAt: new Date(), + }).where(and( + eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), + )).returning({ id: heartbeatRuns.id }); + if (!released) throw new Error("native_run_resources_not_released"); return { effectKind: effect.kind, targetType: "heartbeat_run", - targetId: input.runId, - payload: {}, + targetId: released.id, + payload: { processReleased: true }, }; } - if (effect.kind === "record_expiry" || effect.kind === "schedule_reconciliation") { + if (effect.kind === "record_expiry") { failAt("recovery_materialization", input.failpoint); if (!input.issue.assigneeAgentId) throw new Error("native_recovery_owner_missing"); - const cause = effect.kind === "record_expiry" ? "attention_expired" : "native_reconciliation_required"; + const interaction = await input.tx.select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issue.id), + inArray(issueThreadInteractions.status, ["pending", "expired"]), + )).for("update").limit(1).then((rows) => rows[0] ?? null); + if (!interaction) throw new Error("native_attention_expiry_target_missing"); + await input.tx.update(issueThreadInteractions).set({ + status: "expired", + resolvedAt: new Date(), + updatedAt: new Date(), + }).where(eq(issueThreadInteractions.id, interaction.id)); + const cause = "attention_expired"; const recovery = await issueRecoveryActionService(input.tx).upsertSourceScoped({ companyId: input.companyId, sourceIssueId: input.issue.id, @@ -354,9 +559,31 @@ async function materializeDecisionEffect(input: { cause, fingerprint: nativeSha256({ runId: input.runId, decisionId: input.decisionId, cause }), evidence: { runId: input.runId, decisionId: input.decisionId }, - nextAction: effect.kind === "record_expiry" - ? "Review the expired attention route and choose a new authorized path." - : "Reconcile the native decision against the latest issue status version.", + nextAction: "Review the expired attention route and choose a new authorized path.", + wakePolicy: null, + maxAttempts: 3, + }); + return { + effectKind: effect.kind, + targetType: "issue_thread_interaction", + targetId: interaction.id, + payload: { cause, recoveryActionId: recovery.id }, + }; + } + if (effect.kind === "schedule_reconciliation") { + failAt("recovery_materialization", input.failpoint); + if (!input.issue.assigneeAgentId) throw new Error("native_recovery_owner_missing"); + const cause = "native_reconciliation_required"; + const recovery = await issueRecoveryActionService(input.tx).upsertSourceScoped({ + companyId: input.companyId, + sourceIssueId: input.issue.id, + kind: "active_run_watchdog", + ownerType: "agent", + ownerAgentId: input.issue.assigneeAgentId, + cause, + fingerprint: nativeSha256({ runId: input.runId, decisionId: input.decisionId, cause }), + evidence: { runId: input.runId, decisionId: input.decisionId }, + nextAction: "Reconcile the native decision against the latest issue status version.", wakePolicy: null, maxAttempts: 3, }); @@ -367,6 +594,340 @@ async function materializeDecisionEffect(input: { payload: { cause }, }; } + if (effect.kind === "accept_replacement_turn") { + const [run] = await input.tx.update(heartbeatRuns).set({ + status: "running", + continuationAttempt: sql`${heartbeatRuns.continuationAttempt} + 1`, + nextAction: "Accept a replacement native turn on the existing run.", + updatedAt: new Date(), + }).where(and( + eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), + )).returning({ id: heartbeatRuns.id, continuationAttempt: heartbeatRuns.continuationAttempt }); + if (!run) throw new Error("native_replacement_turn_not_accepted"); + return { + effectKind: effect.kind, + targetType: "heartbeat_run", + targetId: run.id, + payload: { continuationAttempt: run.continuationAttempt }, + }; + } + if (effect.kind === "cancel_continuations") { + const wakeRows = await input.tx.select({ id: agentWakeupRequests.id }) + .from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, input.companyId), + inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution", "claimed"]), + sql`coalesce( + ${agentWakeupRequests.payload} ->> 'issueId', + ${agentWakeupRequests.payload} ->> 'taskId', + ${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'issueId', + ${agentWakeupRequests.payload} -> '_paperclipWakeContext' ->> 'taskId' + ) = ${input.issue.id}`, + )); + if (wakeRows.length > 0) { + await input.tx.update(agentWakeupRequests).set({ + status: "cancelled", + finishedAt: new Date(), + error: "Cancelled by native issue cancellation", + updatedAt: new Date(), + }).where(inArray(agentWakeupRequests.id, wakeRows.map((row) => row.id))); + } + const interactionRows = await input.tx.select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issue.id), + eq(issueThreadInteractions.status, "pending"), + )); + if (interactionRows.length > 0) { + await input.tx.update(issueThreadInteractions).set({ + status: "cancelled", + resolvedAt: new Date(), + updatedAt: new Date(), + }).where(inArray(issueThreadInteractions.id, interactionRows.map((row) => row.id))); + } + const [run] = await input.tx.update(heartbeatRuns).set({ + status: "cancelled", + finishedAt: new Date(), + updatedAt: new Date(), + }).where(and( + eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), + )).returning({ id: heartbeatRuns.id }); + if (!run) throw new Error("native_continuation_cancellation_run_missing"); + return { + effectKind: effect.kind, + targetType: "heartbeat_run", + targetId: run.id, + payload: { + cancelledWakeIds: wakeRows.map((row) => row.id), + cancelledInteractionIds: interactionRows.map((row) => row.id), + }, + }; + } + if (effect.kind === "append_superseding_assessment") { + if (!input.issue.lastStatusDecisionId) throw new Error("native_superseded_decision_missing"); + const [decision] = await input.tx.update(statusDecisions).set({ + supersedesDecisionId: input.issue.lastStatusDecisionId, + }).where(and( + eq(statusDecisions.id, input.decisionId), + eq(statusDecisions.companyId, input.companyId), + )).returning({ id: statusDecisions.id }); + if (!decision) throw new Error("native_superseding_assessment_not_linked"); + return { + effectKind: effect.kind, + targetType: "status_decision", + targetId: decision.id, + payload: { supersedesDecisionId: input.issue.lastStatusDecisionId }, + }; + } + if (effect.kind === "dispatch_pending_effect") { + const pending = await input.tx.select({ id: statusDecisionEffects.id }) + .from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.companyId, input.companyId), + eq(statusDecisionEffects.issueId, input.issue.id), + inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), + )).for("update"); + if (pending.length === 0) throw new Error("native_pending_effect_missing"); + await input.tx.update(statusDecisionEffects).set({ + deliveryState: "delivered", + attemptCount: sql`${statusDecisionEffects.attemptCount} + 1`, + deliveredAt: new Date(), + nextAttemptAt: null, + lastError: null, + updatedAt: new Date(), + }).where(inArray(statusDecisionEffects.id, pending.map((row) => row.id))); + await updateRunEffectState({ + tx: input.tx, + companyId: input.companyId, + runId: input.runId, + key: "nativeDispatchedEffectIds", + value: pending.map((row) => row.id), + }); + return { + effectKind: effect.kind, + targetType: "heartbeat_run", + targetId: input.runId, + payload: { dispatchedEffectIds: pending.map((row) => row.id) }, + }; + } + 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 }); + if (!updated) throw new Error("native_status_version_not_incremented"); + return { + effectKind: effect.kind, + targetType: "issue", + targetId: updated.id, + payload: { statusVersion: updated.statusVersion }, + }; + } + if (effect.kind === "record_shadow_decision") { + const targetId = await updateRunEffectState({ + tx: input.tx, + companyId: input.companyId, + runId: input.runId, + key: "nativeShadowDecision", + value: { decisionId: input.decisionId, applicationState: "shadow" }, + }); + return { + effectKind: effect.kind, + targetType: "heartbeat_run", + targetId, + payload: { decisionId: input.decisionId, applicationState: "shadow" }, + }; + } + if (effect.kind === "render_four_layers") { + const targetId = await updateRunEffectState({ + tx: input.tx, + companyId: input.companyId, + runId: input.runId, + key: "nativeOutcomeLayers", + value: { + runStatus: "persisted", + issueStatus: input.issue.status, + claim: "preserved", + decisionId: input.decisionId, + }, + }); + return { + effectKind: effect.kind, + targetType: "heartbeat_run", + targetId, + payload: { layerCount: 4 }, + }; + } + if (effect.kind === "materialize_contract") { + const contract = await input.tx.select({ id: completionContracts.id, canonicalSha256: completionContracts.canonicalSha256 }) + .from(completionContracts).where(and( + eq(completionContracts.companyId, input.companyId), + eq(completionContracts.issueId, input.issue.id), + )).orderBy(desc(completionContracts.revision)).limit(1).then((rows) => rows[0] ?? null); + if (!contract) throw new Error("native_completion_contract_missing"); + await input.tx.update(heartbeatRuns).set({ + completionContractId: contract.id, + completionContractSha256: contract.canonicalSha256, + updatedAt: new Date(), + }).where(and( + eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), + )); + return { + effectKind: effect.kind, + targetType: "completion_contract", + targetId: contract.id, + payload: { canonicalSha256: contract.canonicalSha256 }, + }; + } + if (effect.kind === "record_mode_labeled_divergence") { + const targetId = await updateRunEffectState({ + tx: input.tx, + companyId: input.companyId, + runId: input.runId, + key: "nativeLegacyDivergence", + value: { mode: "native", decisionId: input.decisionId, classified: true }, + }); + return { + effectKind: effect.kind, + targetType: "heartbeat_run", + targetId, + payload: { mode: "native", classified: true }, + }; + } + if (effect.kind === "record_mode_native") { + const [run] = await input.tx.update(heartbeatRuns).set({ + runtimeMode: "native", + runtimeModeResolvedAt: new Date(), + updatedAt: new Date(), + }).where(and( + eq(heartbeatRuns.id, input.runId), + eq(heartbeatRuns.companyId, input.companyId), + )).returning({ id: heartbeatRuns.id }); + if (!run) throw new Error("native_runtime_mode_not_recorded"); + return { effectKind: effect.kind, targetType: "heartbeat_run", targetId: run.id, payload: { runtimeMode: "native" } }; + } + if (effect.kind === "record_policy_version") { + const targetId = await updateRunEffectState({ + tx: input.tx, + companyId: input.companyId, + runId: input.runId, + key: "nativeStatusPolicyVersion", + value: NATIVE_STATUS_ARBITER_POLICY_VERSION, + runnerProfile: true, + }); + return { + effectKind: effect.kind, + targetType: "heartbeat_run", + targetId, + payload: { policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION }, + }; + } + if (effect.kind === "finish_as_native") { + const targetId = await updateRunEffectState({ + tx: input.tx, + companyId: input.companyId, + runId: input.runId, + key: "nativeKillSwitchDisposition", + value: "finish_as_native", + }); + await input.tx.update(heartbeatRuns).set({ runtimeMode: "native", updatedAt: new Date() }) + .where(eq(heartbeatRuns.id, input.runId)); + 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, + }); + 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"); + return { + effectKind: effect.kind, + targetType: "workspace_operation", + targetId: resumed.id, + payload: { resumedFromOperationId: previous.id }, + }; + } + if (effect.kind === "record_stale_response") { + const interaction = await input.tx.select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issue.id), + inArray(issueThreadInteractions.status, ["answered", "accepted", "rejected", "expired"]), + )).orderBy(desc(issueThreadInteractions.updatedAt)).limit(1).then((rows) => rows[0] ?? null); + if (!interaction) throw new Error("native_stale_response_missing"); + await input.tx.update(issueThreadInteractions).set({ + summary: "Response retained for audit after native supersession.", + updatedAt: new Date(), + }).where(eq(issueThreadInteractions.id, interaction.id)); + return { effectKind: effect.kind, targetType: "issue_thread_interaction", targetId: interaction.id, payload: { status: "superseded" } }; + } + if (effect.kind === "link_canonical_request") { + const interactions = await input.tx.select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issue.id), + )).orderBy(asc(issueThreadInteractions.createdAt)).limit(2); + if (interactions.length < 2) throw new Error("native_attention_canonical_pair_missing"); + const canonical = interactions[0]!; + const duplicate = interactions[1]!; + await input.tx.update(issueThreadInteractions).set({ + summary: `Canonical native attention request: ${canonical.id}`, + updatedAt: new Date(), + }).where(eq(issueThreadInteractions.id, duplicate.id)); + return { + effectKind: effect.kind, + targetType: "issue_thread_interaction", + targetId: duplicate.id, + payload: { canonicalRequestId: canonical.id }, + }; + } + if (effect.kind === "release_checkout") { + const [released] = await input.tx.update(issues).set({ + checkoutRunId: null, + executionRunId: null, + executionAgentNameKey: null, + executionLockedAt: null, + updatedAt: new Date(), + }).where(and( + eq(issues.id, input.issue.id), + eq(issues.companyId, input.companyId), + )).returning({ id: issues.id }); + if (!released) throw new Error("native_checkout_not_released"); + return { + effectKind: effect.kind, + targetType: "issue_checkout", + targetId: released.id, + payload: { checkoutRunId: null, executionRunId: null }, + }; + } if (effect.kind === "record_recovery") { failAt("recovery_materialization", input.failpoint); const recovery = await issueRecoveryActionService(input.tx).upsertSourceScoped({ @@ -389,12 +950,7 @@ async function materializeDecisionEffect(input: { payload: { cause: effect.cause, nextAction: effect.nextAction }, }; } - return { - effectKind: effect.kind, - targetType: "issue_checkout", - targetId: input.issue.id, - payload: { checkoutRunId: input.issue.checkoutRunId, executionRunId: input.issue.executionRunId }, - }; + return assertNeverEffect(effect); } export async function commitNativeStatusDecision(input: { @@ -425,15 +981,25 @@ export async function commitNativeStatusDecision(input: { eq(issues.id, input.issueId), eq(issues.companyId, input.companyId), )).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 ( - !issue - || issue.status !== input.priorStatus + issue.status !== input.priorStatus || Number(issue.statusVersion) !== input.priorStatusVersion || issue.lastStatusDecisionId !== input.priorDecisionId ) { throw new NativeStatusRaceError(); } const decisionJson = { + statusAction: input.decision.statusAction, toStatus: input.decision.toStatus, reasonCode, unblockDescriptor: input.decision.unblockDescriptor, @@ -452,7 +1018,7 @@ export async function commitNativeStatusDecision(input: { eq(statusDecisions.issueId, input.issueId), eq(statusDecisions.decisionDigest, decisionDigest), )).limit(1).then((rows) => rows[0] ?? null); - if (decisionRow?.applicationState === "applied") { + if (decisionRow?.applicationState === "applied" || decisionRow?.applicationState === "proposed") { return { decision: decisionRow, issue, replayed: true }; } if (!decisionRow) { @@ -485,24 +1051,34 @@ export async function commitNativeStatusDecision(input: { })); } - failAt("status_projection", input.failpoint); - const updated = await issueService(tx as unknown as Db).update(input.issueId, { - status: input.decision.toStatus, - statusVersion: input.priorStatusVersion + 1, - lastStatusDecisionId: decisionRow.id, - unblockDescriptor: input.decision.unblockDescriptor, - actorAgentId: null, - actorUserId: null, - }, tx, publications); - if (!updated) throw new NativeStatusRaceError(); - materialized.unshift({ - effectKind: "issue_status_projection", - targetType: "issue", - targetId: input.issueId, - payload: { fromStatus: issue.status, toStatus: input.decision.toStatus, reasonCode }, - }); + let updated: typeof issues.$inferSelect; + if (input.decision.statusAction === "preserve") { + updated = await tx.select().from(issues).where(and( + eq(issues.id, input.issueId), + eq(issues.companyId, input.companyId), + )).limit(1).then((rows) => rows[0] ?? null) as typeof issues.$inferSelect; + if (!updated) throw new NativeStatusRaceError(); + } else { + failAt("status_projection", input.failpoint); + const projected = await issueService(tx as unknown as Db).update(input.issueId, { + status: input.decision.toStatus, + statusVersion: input.priorStatusVersion + 1, + lastStatusDecisionId: decisionRow.id, + unblockDescriptor: input.decision.unblockDescriptor, + actorAgentId: null, + actorUserId: null, + }, tx, publications); + if (!projected) throw new NativeStatusRaceError(); + updated = projected; + materialized.unshift({ + effectKind: "issue_status_projection", + targetType: "issue", + targetId: input.issueId, + payload: { fromStatus: issue.status, toStatus: input.decision.toStatus, reasonCode }, + }); + } - if (input.decision.toStatus === "done" && issue.status !== "done") { + if (input.decision.statusAction === "done" && issue.status !== "done") { const issueSvc = issueService(tx as unknown as Db); const dependents = await issueSvc.listWakeableBlockedDependents(input.issueId); for (const dependent of dependents) { @@ -564,24 +1140,34 @@ export async function commitNativeStatusDecision(input: { deliveredAt: new Date(), }).onConflictDoNothing(); } - await tx.update(statusDecisions).set({ applicationState: "applied", appliedAt: new Date() }) + const shadowOnly = input.decision.effects.some((effect) => effect.kind === "record_shadow_decision"); + const finalizationError = input.decision.effects.find((effect) => effect.kind === "record_finalization_error"); + const applicationState = shadowOnly ? "proposed" : "applied"; + await tx.update(statusDecisions).set({ + applicationState, + appliedAt: shadowOnly ? null : new Date(), + }) .where(eq(statusDecisions.id, decisionRow.id)); await tx.update(nativeRunFinalizations).set({ - phase: "committed", + phase: finalizationError ? "retryable_failure" : "committed", assessmentId: input.assessmentId, decisionId: decisionRow.id, leaseOwner: null, leaseExpiresAt: null, - failureCode: null, - failureDetail: null, - nextAttemptAt: null, + failureCode: finalizationError?.cause ?? null, + failureDetail: finalizationError ? { + originalFailureCode: finalizationError.cause, + recoveryOwner: { kind: "agent", agentId: finalizationError.agentId }, + nextAction: finalizationError.nextAction, + } : null, + nextAttemptAt: finalizationError ? new Date(Date.now() + 30_000) : null, updatedAt: new Date(), }).where(eq(nativeRunFinalizations.runId, input.runId)); const { publication } = await persistActivity(tx as unknown as Db, { companyId: input.companyId, actorType: "system", actorId: "native-status-committer", - action: "issue.updated", + action: input.decision.statusAction === "preserve" ? "issue.status_decision_recorded" : "issue.updated", entityType: "issue", entityId: input.issueId, issueId: input.issueId, @@ -598,7 +1184,7 @@ export async function commitNativeStatusDecision(input: { }); publications.push(publication); return { - decision: { ...decisionRow, applicationState: "applied" }, + decision: { ...decisionRow, applicationState }, issue: updated, replayed: false, };