fix(native-runner): materialize phase 6 live effects

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-09 06:05:21 +00:00
1 parent 33d9e8ef66
commit 3bed55f3f7
14 files changed
+1920 -478

No files matched your search

@@ -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
@@ -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
@@ -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.
@@ -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.
@@ -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,
@@ -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<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", "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<NativeStatusEffect["kind"]>([
"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<string>();
const priorIssueStatus = String(fixture.given.priorIssueStatus ?? "in_progress") as Parameters<typeof arbitrateNativeStatus>[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<typeof resolveNativeFinalizerStatus>,
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<typeof resolveNativeFinalizerStatus> | 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 });
});
});
+2 -2
View File
@@ -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,
@@ -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<NativeStatusAuthorityScenario>([
"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,
@@ -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<string, unknown> {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: {};
}
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<NativeAttentionMaterializedTarget[]> {
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<NativeInteractionResponseEnvelope[]> {
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<NativeStatusAuthorityScenario>([
"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");
}
@@ -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<typeof arbitrateNativeStatus>[0]) {
return "scenario" in input
? arbitrateNativeStatusScenario(input)
: arbitrateNativeStatus(input);
/** Fact-based arbitration at the production finalization authority seam. */
export function resolveNativeFinalizerStatus(input: Parameters<typeof arbitrateNativeStatus>[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;
@@ -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<boolean> {
function record(value: unknown): Record<string, unknown> {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: {};
}
export function cancelNativeSession(runId: string, reason: string): Promise<boolean>;
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<boolean | { dispatched: boolean; decision: NativeStatusDecision | null }> {
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: {
@@ -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<string, unknown> {
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<NativeStatusAuthorityScenario>([
"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<NativeStatusAuthorityScenario>([
"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");
}
@@ -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;
@@ -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<string, unknown>;
};
function record(value: unknown): Record<string, unknown> {
return value && typeof value === "object" && !Array.isArray(value)
? value as Record<string, unknown>
: {};
}
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,
};