fix(native-runner): prove live Phase 6 effects

Route corpus operations through cancellation, reconciliation, attention, and rollout entrypoints, materialize effects through owning services, and document the live-versus-policy proof boundary.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-08-09 06:46:53 +00:00
1 parent 3bed55f3f7
commit d53b7902ab
13 files changed
+1272 -168

No files matched your search

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