mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-09 07:10:34 +01:00
Settle saved monitor results after a newer user request (#19644)
## Thinking Path
> - Paperclip is the open source app people use to manage AI agents for
work.
> - Native results must finish their durable finalization before later
work can reuse a saved workspace.
> - An accepted result can wait on a monitor while its workspace files
are recovered.
> - A newer user request can replace that old wait before finalization
resumes.
> - The old finalizer then reports lost monitor authority and blocks the
saved-workspace lineage.
> - This pull request proves the newer user request and settles only the
old result bookkeeping.
> - The benefit is released recovery work without replaying an old turn
or changing the current request.
## Linked Issues or Issue Description
Refs #17641 and #16975.
**What happened?**
An older successful native result has saved files and a server-issued
monitor receipt. Its task is now blocked after a newer user-comment
continuation failed during preparation. The old coordinator stays in
retryable_failure with monitor_wait_authority_lost. The retained
allocation ownership check correctly refuses the unresolved predecessor,
so later file recovery also stays held.
**Expected behavior**
The normal finalizer can settle the old accepted result when exact newer
user-request evidence proves the old wait is superseded. It must
preserve the current task, user input, ownership, schedule, accepted
result, and attempt history. It must not start provider work or publish
the old wait summary.
**Steps to reproduce**
Create an accepted successful monitor result and mark workspace
finalization successful. Leave the task blocked. Record a newer
canonical user comment and its ordinary user wake, either queued or
consumed by a same-task continuation that retains the exact comment
revision and source-run lineage. Call the real native finalizer. Before
the repair, both cases remain retryable_failure instead of settling.
**Paperclip version or commit**
e482f0f632, before this repair.
## What Changed
- Add a company-scoped passive supersession proof in the native monitor
owner. It applies only when the old wait is not currently eligible. The
proof binds the old accepted result and monitor receipt to a newer
canonical unedited user comment and matching user wake. Admitted
successors must retain the exact task, comment revision, and source
lineage.
- Recheck the proof under the existing task transaction, with
nonblocking shared locks on the comment and wake; contention uses the
existing status-race retry path. Commit a preserve decision with no
effects. Keep normal monitor authority, Stop, workspace finalization,
and retained-allocation guards unchanged.
- Add real PostgreSQL coverage for queued and consumed input, current
successor ownership, newer schedules, immutable accepted results,
attempt history, foreign/missing/cancelled evidence, and changes before
commit. Real scheduler ticks create no old monitor wake.
- Include the present superseding proof in the existing assessment
identity. The old failed assessment and decision stay immutable, while
the new decision gets its own assessment. Absent proof keeps the
existing fingerprint unchanged.
- Document passive finalization after a newer user request.
## Verification
- Behavioral RED before production edits: two real PostgreSQL finalizer
cases returned monitor_wait_authority_lost/retryable_failure instead of
settling after a newer queued or consumed user request. Initial
interrupted imports and rejected database setup were not counted as
behavioral failures.
- Additional real-boundary RED: the normal finalization reconciler
reconstructed both persisted errors but failed on
status_decisions_company_assessment_uq because the changed proof reused
an older assessment identity. The owned PostgreSQL log confirmed the
constraint. Including only the present proof in the existing fingerprint
makes both reconciler cases pass while old assessments and decisions
remain immutable.
- Review RED: real comment-row and wake-row contention produced
side_effect_planning_failed. A test-local 100ms lock timeout bounds this
diagnostic. Nonblocking proof locks and `55P03`→NativeStatusRaceError
safely defer both cases, preserve current state, and commit after the
original lock owner releases. Both regressions pass without weakening
the production lock or authority checks.
- Final local GREEN at `0c40331d8d7f587c9a532fbc51d358bfacd1bd46`: five
files, 111 passing tests (126.17s), including all 54 saved-monitor
cases, actual persisted-error reconciliation with an absent retry
deadline and older boot identity, the status
arbiter/task-monitor/terminal-status suites, and six model fixtures.
Current requests, schedules, holds, accepted results and prior
assessments/decisions remain unchanged; normal attempt increments and
idempotent repeat are asserted. Real scheduler ticks create no old
monitor wake.
- All seven server typecheck layers, module-boundary and whitespace
checks pass at the final inputs. Required preflight passes in 15.348s.
No new code comments were added. Local environment: macOS arm64, Node
26.4.0, pinned pnpm 9.15.4, Vitest 5.0.3, owned disposable PostgreSQL 18
with bounded 1GiB tmpfs; no shared Docker data was pruned.
- [Final-head
CI](https://github.com/paperclipai/paperclip/actions/runs/42311889918):
59 successful checks,three intentional skips,one neutral optional
review,zero failed or pending checks. Required E2E and verify aggregates
pass. Greptile reviewed this exact head at 5/5 with no unresolved
findings.
- The [original support
job](https://github.com/paperclipai/paperclip/actions/runs/42311889918/job/128217026059)
passed 6849 assertions and hit one 30000ms timeout in an unchanged
private-evidence fixture. The exact case passed locally; all 41 cases
passed together (42.79s). The [identical-input sibling CI
control](https://github.com/paperclipai/paperclip/actions/runs/42311878548/job/128216598362)
passed all 6850 assertions and the 41-case file in 5.417s. Exactly one
targeted replay was authorized and dispatched after workflow completion
and ordinary quota reset, preserving the other 57 passing checks.
[Replay
job128227417618](https://github.com/paperclipai/paperclip/actions/runs/42311889918/job/128227417618)
passed 323 files/6850 assertions with four existing skips; the 41-case
file passed in 5.665s. No fixture, source, assertion, timeout,
credentials or full workflow was changed/replayed. The original timeout
remains recorded; its precise host-level cause was not determined.
- Inherited baseline failures were handled through normal integration of
existing #19610 (model expectations), #19627 (deterministic Slack
deadline fixture), and #19646 (current trusted Runner lane flags). All
four exact lane-command dry runs and the lane-inventory regression pass.
No force push or new production patch was used for those failures.
- Hosted recovery remains pending deployment and acceptance through the
normal pipeline; these checks do not claim the live task has already
recovered.
## Risks
- The new proof must remain passive. It cannot grant execution or
restore an obsolete task state. Missing or changing proof leaves the old
result held.
- The query reads only the same company's task and agent, with requests
newer than the old completion time and one selected result. It runs
during active finalization and adds no timer or idle polling.
- Edited comments, cancelled wakes, foreign lineage, source Stop, and
workspace failure are refused. Existing workspace snapshot, stop, and
handoff checks still own physical file recovery.
- Hosted acceptance remains pending deployment through the normal
pipeline.
## Model Used
- OpenAI Codex (GPT-6 family), with code execution and repository tools.
The runtime does not expose the exact serving model identifier or
context window size.
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
---------
Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
e33aa8a46a
commit
56c2b28c04
7 files changed
+316
-8
No files matched your search
@@ -774,6 +774,19 @@ durable wake and an admitted successor run. The old run must no longer own the
|
||||
task's checkout or execution. This settlement preserves the task and successor
|
||||
ownership, and neither re-arms the timer nor starts another provider turn.
|
||||
Workspace persistence must still succeed first.
|
||||
|
||||
When the old monitor wait is no longer eligible, a newer ordinary user-comment
|
||||
wake can supersede that old saved wait.
|
||||
The finalizer requires the canonical unedited user comment, the matching user wake,
|
||||
and the accepted old result with its server-issued monitor receipt. An admitted
|
||||
successor must retain the exact task, comment revision, and source-run lineage.
|
||||
The finalizer rechecks this proof under the task lock. Assessment fingerprints
|
||||
include the exact superseding request proof, so a prior failed assessment remains
|
||||
immutable when the finalization decision changes. It records a passive
|
||||
settlement and preserves the current task status, assignment, checkout, schedule,
|
||||
and newer request. It does not re-arm the old monitor, publish its old waiting
|
||||
summary, or start another provider turn. Missing, cancelled, foreign, or conflicting
|
||||
proof keeps the old result held. Source Stop and workspace persistence gates remain.
|
||||
If a saved monitor expires during workspace recovery, settlement releases the
|
||||
old run into the still-persisted schedule. The scheduler then applies its timeout
|
||||
or attempt-limit recovery policy. Expiration still rejects new monitor-wait
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { agentWakeupRequests, heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db, type issues } from "@paperclipai/db";
|
||||
import { and, desc, eq, inArray, isNotNull, ne, sql } from "drizzle-orm";
|
||||
import { agentWakeupRequests, heartbeatRuns, issueComments, nativeRunFinalizations, nativeRunResults, type Db, type issues } from "@paperclipai/db";
|
||||
import { and, desc, eq, gt, inArray, isNotNull, isNull, ne, notInArray, or, sql } from "drizzle-orm";
|
||||
import { alias } from "drizzle-orm/pg-core";
|
||||
import { boundRunIssueIdSql } from "../run-task-binding.js";
|
||||
import { scheduledIssueMonitorWait } from "../issue-monitors.js";
|
||||
@@ -21,7 +21,14 @@ export async function readNativePreservedMonitorWait(
|
||||
issue.executionRunId !== null || issue.checkoutRunId !== null) return null;
|
||||
const nextCheckAt = scheduledIssueMonitorWait(issue, binding.agentId);
|
||||
if (!nextCheckAt) return null;
|
||||
const [source] = await db.select({ resultId: nativeRunResults.id, hasReceipt: sql<boolean>`exists (
|
||||
const source = await readCompletedNativeMonitorResult(db, binding);
|
||||
return source?.hasReceipt ? { agentId: binding.agentId, nextCheckAt, resultId: source.resultId } : null;
|
||||
}
|
||||
|
||||
type MonitorBinding = { companyId: string; issueId: string; runId: string; agentId: string };
|
||||
|
||||
async function readCompletedNativeMonitorResult(db: Db, binding: MonitorBinding) {
|
||||
const [source] = await db.select({ resultId: nativeRunResults.id, finishedAt: heartbeatRuns.finishedAt, hasReceipt: sql<boolean>`exists (
|
||||
select 1 from jsonb_each(case when jsonb_typeof(${heartbeatRuns.resultJson}->'semanticToolReceipts') = 'object'
|
||||
then ${heartbeatRuns.resultJson}->'semanticToolReceipts' else '{}'::jsonb end) receipt
|
||||
where receipt.value->>'operationId' = 'set_task_monitor'
|
||||
@@ -41,7 +48,54 @@ export async function readNativePreservedMonitorWait(
|
||||
...["cancellation", "nativeCancellation", "executionCancellation", "startupCancellation"].map(key =>
|
||||
sql`coalesce(${heartbeatRuns.resultJson}->${key}, 'null'::jsonb) = 'null'::jsonb`),
|
||||
)).limit(1);
|
||||
return source?.hasReceipt ? { agentId: binding.agentId, nextCheckAt, resultId: source.resultId } : null;
|
||||
return source?.hasReceipt ? source : null;
|
||||
}
|
||||
|
||||
export type NativeSupersededMonitorWait = {
|
||||
agentId: string; resultId: string; wakeId: string; commentId: string; successorRunId: string | null;
|
||||
};
|
||||
|
||||
export async function readNativeSupersededMonitorWait(db: Db, binding: MonitorBinding,
|
||||
issue: typeof issues.$inferSelect, lockProof = false): Promise<NativeSupersededMonitorWait | null> {
|
||||
if (issue.companyId !== binding.companyId || issue.id !== binding.issueId ||
|
||||
issue.executionRunId === binding.runId || issue.checkoutRunId === binding.runId ||
|
||||
scheduledIssueMonitorWait(issue, binding.agentId) !== null) return null;
|
||||
const source = await readCompletedNativeMonitorResult(db, binding);
|
||||
if (!source?.finishedAt) return null;
|
||||
const successor = alias(heartbeatRuns, "monitor_superseding_run");
|
||||
const query = db.select({ wakeId: agentWakeupRequests.id, commentId: issueComments.id,
|
||||
successorRunId: successor.id }).from(agentWakeupRequests).innerJoin(issueComments, and(
|
||||
eq(issueComments.companyId, binding.companyId), eq(issueComments.issueId, binding.issueId),
|
||||
eq(issueComments.authorType, "user"), eq(issueComments.authorUserId, agentWakeupRequests.requestedByActorId),
|
||||
isNull(issueComments.authorAgentId), isNull(issueComments.createdByRunId),
|
||||
isNull(issueComments.deletedAt), isNull(issueComments.sourceTrust),
|
||||
gt(issueComments.createdAt, source.finishedAt),
|
||||
sql`${issueComments.updatedAt} = ${issueComments.createdAt}`,
|
||||
sql`length(trim(${issueComments.body})) > 0`,
|
||||
sql`${agentWakeupRequests.payload}->>'commentId' = ${issueComments.id}::text`,
|
||||
)).leftJoin(successor, and(eq(successor.id, agentWakeupRequests.runId),
|
||||
eq(successor.companyId, binding.companyId), eq(successor.agentId, binding.agentId),
|
||||
eq(successor.nativeIssueId, binding.issueId), ne(successor.id, binding.runId),
|
||||
eq(successor.wakeupRequestId, agentWakeupRequests.id),
|
||||
sql`${successor.contextSnapshot}#>>'{executionContinuation,companyId}' = ${binding.companyId}`,
|
||||
sql`${successor.contextSnapshot}#>>'{executionContinuation,issueId}' = ${binding.issueId}`,
|
||||
sql`${successor.contextSnapshot}#>>'{executionContinuation,resumeDelta,baseRunId}' = ${binding.runId}`,
|
||||
sql`${successor.contextSnapshot}#>>'{executionContinuation,objectiveSource,kind}' = 'comment'`,
|
||||
sql`${successor.contextSnapshot}#>>'{executionContinuation,objectiveSource,id}' = ${issueComments.id}::text`,
|
||||
sql`${successor.contextSnapshot}#>>'{executionContinuation,objectiveSource,revision}' =
|
||||
to_char(${issueComments.updatedAt} at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')`,
|
||||
)).where(and(eq(agentWakeupRequests.companyId, binding.companyId),
|
||||
eq(agentWakeupRequests.agentId, binding.agentId), gt(agentWakeupRequests.requestedAt, source.finishedAt),
|
||||
inArray(agentWakeupRequests.source, ["automation", "on_demand"]),
|
||||
inArray(agentWakeupRequests.reason, ["issue_commented", "issue_reopened_via_comment"]),
|
||||
eq(agentWakeupRequests.requestedByActorType, "user"),
|
||||
sql`${agentWakeupRequests.payload}->>'issueId' = ${binding.issueId}`,
|
||||
notInArray(agentWakeupRequests.status, ["skipped", "cancelled"]),
|
||||
or(and(isNull(agentWakeupRequests.runId), inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"])),
|
||||
isNotNull(successor.id)),
|
||||
)).orderBy(desc(issueComments.createdAt), desc(issueComments.id)).limit(1);
|
||||
const [request] = await (lockProof ? query.for("share", { of: [agentWakeupRequests, issueComments], noWait: true }) : query);
|
||||
return request ? { ...request, agentId: binding.agentId, resultId: source.resultId } : null;
|
||||
}
|
||||
|
||||
export async function readNativeMonitorSource(
|
||||
|
||||
@@ -2,7 +2,7 @@ import { randomUUID } from "node:crypto";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { agents, agentWakeupRequests, companies, completionContracts, createDb, applyPendingMigrations, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns,
|
||||
issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, subscribeDatabaseWork } from "@paperclipai/db";
|
||||
issueComments, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, subscribeDatabaseWork, workspaceOperations, workAssessments } from "@paperclipai/db";
|
||||
import { PROVIDER_QUOTA_MONITOR_SERVICE_NAME } from "@paperclipai/shared";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
|
||||
@@ -13,6 +13,8 @@ import * as monitorDelivery from "./native-monitor-delivery.js";
|
||||
import { DELIVERY_QUEUES } from "../delivery-work-notifications.js";
|
||||
import { heartbeatService } from "../heartbeat.js";
|
||||
import { hasPendingRunCleanup } from "../execution-blocker.js";
|
||||
import { reconcileNativeFinalizations } from "./native-finalization-reconciler.js";
|
||||
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
|
||||
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_PRESERVED_MONITOR_TEST_DATABASE_URL");
|
||||
const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
@@ -221,6 +223,210 @@ describe.skipIf(!support.supported)("released native monitor results", () => {
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([]);
|
||||
});
|
||||
|
||||
async function newerUserRequest(state: "queued" | "consumed", existing?: Awaited<ReturnType<typeof fixture>>) {
|
||||
const f = existing ?? await fixture();
|
||||
if (!existing) await db.update(issues).set({ status: "blocked", statusVersion: 5 }).where(eq(issues.id, f.issueId));
|
||||
const [comment] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId,
|
||||
authorType: "user", authorUserId: "operator", body: "Continue with the new objective." }).returning();
|
||||
const successorId = state === "consumed" ? randomUUID() : null;
|
||||
if (successorId) await db.insert(heartbeatRuns).values({ id: successorId, companyId: f.companyId, agentId: f.agentId,
|
||||
nativeIssueId: f.issueId, runtimeMode: "native", status: "failed", errorCode: "native_continuation_requires_reconciliation",
|
||||
contextSnapshot: { issueId: f.issueId, commentId: comment!.id, wakeCommentId: comment!.id,
|
||||
executionContinuation: { companyId: f.companyId, issueId: f.issueId,
|
||||
objectiveSource: { kind: "comment", id: comment!.id, revision: comment!.updatedAt.toISOString() },
|
||||
resumeDelta: { baseRunId: f.runId } } },
|
||||
resultJson: { executionRecovery: { kind: "bootstrap", providerWorkStarted: false } },
|
||||
});
|
||||
const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
source: "automation", reason: "issue_reopened_via_comment", requestedByActorType: "user", requestedByActorId: "operator",
|
||||
status: successorId ? "failed" : "queued", runId: successorId,
|
||||
payload: { issueId: f.issueId, commentId: comment!.id,
|
||||
_paperclipWakeContext: { issueId: f.issueId, wakeCommentId: comment!.id, wakeCommentIds: [comment!.id] } },
|
||||
}).returning();
|
||||
if (successorId) await db.update(heartbeatRuns).set({ wakeupRequestId: wake!.id }).where(eq(heartbeatRuns.id, successorId));
|
||||
if (!existing) await db.update(nativeRunFinalizations).set({ attempt: 1 }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
return { f, comment: comment!, wake: wake!, successorId };
|
||||
}
|
||||
|
||||
it.each(["queued", "consumed", "owned_successor", "new_monitor"] as const)("settles the old saved monitor result without changing a newer user request (%s)", async state => {
|
||||
const { f, comment, wake, successorId } = await newerUserRequest(state === "queued" ? "queued" : "consumed");
|
||||
if (state === "owned_successor") {
|
||||
await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, successorId!));
|
||||
await db.update(issues).set({ executionRunId: successorId, checkoutRunId: successorId }).where(eq(issues.id, f.issueId));
|
||||
}
|
||||
if (state === "new_monitor") {
|
||||
const task = await f.readIssue(), nextCheckAt = new Date(Date.now() + 7_200_000);
|
||||
await db.update(issues).set({ monitorNextCheckAt: nextCheckAt, executionPolicy: { ...task.executionPolicy,
|
||||
monitor: { ...(task.executionPolicy?.monitor as Record<string, unknown>), nextCheckAt: nextCheckAt.toISOString() } } })
|
||||
.where(eq(issues.id, f.issueId));
|
||||
}
|
||||
expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(true);
|
||||
const before = await f.readIssue();
|
||||
await f.finalize();
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(coordinator).toMatchObject({ phase: "committed", nextAttemptAt: null, failureCode: null, attempt: 2 });
|
||||
expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(false);
|
||||
expect(await heartbeatService(db, { runtimeEnv: {} }).tickIssueMonitors(new Date(Date.parse(f.nextCheckAt) + 1)))
|
||||
.toMatchObject({ triggered: 0 });
|
||||
expect(await f.readIssue()).toMatchObject({ status: before.status, statusVersion: before.statusVersion,
|
||||
assigneeAgentId: before.assigneeAgentId, assigneeUserId: before.assigneeUserId,
|
||||
executionRunId: before.executionRunId, checkoutRunId: before.checkoutRunId,
|
||||
executionPolicy: before.executionPolicy, monitorNextCheckAt: before.monitorNextCheckAt,
|
||||
monitorAttemptCount: before.monitorAttemptCount, workMode: before.workMode });
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([wake]);
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.id, comment!.id))).toEqual([comment]);
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].resultJson).toEqual(f.result);
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.runId))).toEqual([]);
|
||||
expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0])
|
||||
.toMatchObject({ status: "succeeded", finishedAt: f.finishedAt });
|
||||
const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId));
|
||||
expect(decisions.at(-1)?.decisionJson).toMatchObject({ effects: [] });
|
||||
await f.finalize();
|
||||
expect(await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId))).toEqual(decisions);
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].attempt).toBe(2);
|
||||
});
|
||||
|
||||
it.each(["queued", "consumed"] as const)("reconstructs and settles a historical monitor error through the real finalization reconciler (%s)", async state => {
|
||||
const f = await fixture();
|
||||
await db.update(issues).set({ status: "blocked", statusVersion: 5 }).where(eq(issues.id, f.issueId));
|
||||
await f.finalize();
|
||||
const [prior] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(prior).toMatchObject({ phase: "retryable_failure", failureCode: "monitor_wait_authority_lost" });
|
||||
const oldAssessment = await db.select().from(workAssessments).where(eq(workAssessments.id, prior.assessmentId!));
|
||||
const oldDecision = await db.select().from(statusDecisions).where(eq(statusDecisions.id, prior.decisionId!));
|
||||
expect(oldAssessment).toHaveLength(1);
|
||||
expect(oldDecision).toHaveLength(1);
|
||||
const { comment, wake, successorId } = await newerUserRequest(state, f);
|
||||
let hold: typeof issueRecoveryActions.$inferSelect | undefined;
|
||||
if (successorId) {
|
||||
const action = await issueRecoveryActionService(db).upsertSourceScoped({ companyId: f.companyId, sourceIssueId: f.issueId,
|
||||
kind: "active_run_watchdog", cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(),
|
||||
ownerType: "board", nextAction: "Recover the current user request.", evidence: { runId: successorId }, supersedeOnIdentityChange: true });
|
||||
[hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, action.id));
|
||||
}
|
||||
await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, heartbeatRunId: f.runId,
|
||||
phase: "workspace_finalize", status: "succeeded", finishedAt: new Date() });
|
||||
await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null,
|
||||
controllerBootId: randomUUID() }).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
const before = await f.readIssue();
|
||||
const results = await reconcileNativeFinalizations(db, [f.runId]);
|
||||
expect(results).toContainEqual(expect.objectContaining({ runId: f.runId, phase: "committed" }));
|
||||
const [settled] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(settled).toMatchObject({ phase: "committed", failureCode: null, nextAttemptAt: null });
|
||||
expect(settled.assessmentId).not.toBe(prior.assessmentId);
|
||||
const [assessment] = await db.select().from(workAssessments).where(eq(workAssessments.id, settled.assessmentId!));
|
||||
expect(assessment).toMatchObject({ supersedesAssessmentId: prior.assessmentId, resultId: f.resultId,
|
||||
assessmentJson: { supersededMonitorWait: { agentId: f.agentId, resultId: f.resultId,
|
||||
commentId: comment.id, wakeId: wake.id, successorRunId: successorId } } });
|
||||
expect(await db.select().from(workAssessments).where(eq(workAssessments.id, prior.assessmentId!))).toEqual(oldAssessment);
|
||||
expect(await db.select().from(statusDecisions).where(eq(statusDecisions.id, prior.decisionId!))).toEqual(oldDecision);
|
||||
expect(await hasPendingRunCleanup(db, f.companyId, f.runId)).toBe(false);
|
||||
expect(await f.readIssue()).toMatchObject({ status: before.status, statusVersion: before.statusVersion,
|
||||
executionPolicy: before.executionPolicy, monitorNextCheckAt: before.monitorNextCheckAt,
|
||||
executionRunId: before.executionRunId, checkoutRunId: before.checkoutRunId });
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toEqual([wake]);
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.id, comment.id))).toEqual([comment]);
|
||||
if (hold) expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id))).toEqual([hold]);
|
||||
const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId));
|
||||
expect(decisions.at(-1)?.toStatus).toBe(before.status);
|
||||
expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, f.resultId)))[0].resultJson).toEqual(f.result);
|
||||
});
|
||||
|
||||
it.each(["missing_wake", "actor", "foreign_task", "foreign_company", "untrusted_comment", "deleted_comment", "edited_comment",
|
||||
"prior_comment", "foreign_run", "foreign_base", "wrong_objective", "cancelled_request", "source_stop", "missing_monitor_receipt",
|
||||
"workspace_failed"])("keeps the old monitor held without exact newer user-request proof (%s)", async state => {
|
||||
const consumed = ["foreign_run", "foreign_base", "wrong_objective"].includes(state);
|
||||
const { f, comment, wake, successorId } = await newerUserRequest(consumed ? "consumed" : "queued");
|
||||
if (state === "missing_wake") await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id));
|
||||
if (state === "actor") await db.update(agentWakeupRequests).set({ requestedByActorType: "agent" }).where(eq(agentWakeupRequests.id, wake.id));
|
||||
if (state === "foreign_task") await db.update(agentWakeupRequests).set({ payload: { ...wake.payload, issueId: randomUUID() } })
|
||||
.where(eq(agentWakeupRequests.id, wake.id));
|
||||
if (state === "foreign_company") {
|
||||
const other = await fixture();
|
||||
await db.update(agentWakeupRequests).set({ companyId: other.companyId, agentId: other.agentId })
|
||||
.where(eq(agentWakeupRequests.id, wake.id));
|
||||
}
|
||||
if (state === "untrusted_comment") await db.update(issueComments).set({ sourceTrust: { preset: "low_trust_review", disposition: "quarantined" } }).where(eq(issueComments.id, comment.id));
|
||||
if (state === "deleted_comment") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comment.id));
|
||||
if (state === "edited_comment") await db.update(issueComments).set({ updatedAt: new Date(comment.createdAt.getTime() + 1) }).where(eq(issueComments.id, comment.id));
|
||||
if (state === "prior_comment") await db.update(issueComments).set({ createdAt: new Date(f.finishedAt.getTime() - 1),
|
||||
updatedAt: new Date(f.finishedAt.getTime() - 1) }).where(eq(issueComments.id, comment.id));
|
||||
if (state === "foreign_run") {
|
||||
const [otherIssue] = await db.insert(issues).values({ companyId: f.companyId, title: "Other task", status: "blocked" }).returning();
|
||||
const [otherRun] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
nativeIssueId: otherIssue!.id, runtimeMode: "native", status: "failed", wakeupRequestId: wake.id,
|
||||
contextSnapshot: { issueId: otherIssue!.id } }).returning();
|
||||
await db.update(agentWakeupRequests).set({ runId: otherRun!.id }).where(eq(agentWakeupRequests.id, wake.id));
|
||||
}
|
||||
if (state === "foreign_base" || state === "wrong_objective") await db.update(heartbeatRuns).set({ contextSnapshot:
|
||||
sql`jsonb_set(${heartbeatRuns.contextSnapshot}, ${state === "foreign_base" ? "{executionContinuation,resumeDelta,baseRunId}" : "{executionContinuation,objectiveSource,id}"}::text[], ${JSON.stringify(randomUUID())}::jsonb)` })
|
||||
.where(eq(heartbeatRuns.id, successorId!));
|
||||
if (state === "cancelled_request") await db.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, wake.id));
|
||||
if (state === "source_stop") await db.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson}, '{nativeCancellation}', '{"state":"pending"}'::jsonb)` })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
if (state === "missing_monitor_receipt") await db.update(heartbeatRuns).set({ resultJson: {} }).where(eq(heartbeatRuns.id, f.runId));
|
||||
const before = await f.readIssue();
|
||||
await f.finalize(state === "workspace_failed" ? "failed" : "succeeded");
|
||||
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId));
|
||||
expect(coordinator.phase).toBe("retryable_failure");
|
||||
expect(await f.readIssue()).toMatchObject({ status: before.status, executionPolicy: before.executionPolicy,
|
||||
monitorNextCheckAt: before.monitorNextCheckAt, executionRunId: before.executionRunId, checkoutRunId: before.checkoutRunId });
|
||||
expect(await db.select().from(statusDecisions).where(and(eq(statusDecisions.runId, f.runId),
|
||||
eq(statusDecisions.reasonCode, "scheduled_monitor_superseded")))).toEqual([]);
|
||||
});
|
||||
|
||||
it.each(["comment", "wake"] as const)("defers superseded monitor settlement during child-row contention (%s)", async state => {
|
||||
const { f, comment, wake } = await newerUserRequest("queued");
|
||||
const before = await f.readIssue();
|
||||
let release!: () => void, locked!: () => void;
|
||||
const released = new Promise<void>(resolve => { release = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const owner = db.transaction(async tx => {
|
||||
if (state === "comment") await tx.select().from(issueComments).where(eq(issueComments.id, comment.id)).for("update");
|
||||
else await tx.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id)).for("update");
|
||||
locked();
|
||||
await released;
|
||||
});
|
||||
await acquired;
|
||||
const original = monitorDelivery.readNativeSupersededMonitorWait;
|
||||
const bounded = vi.spyOn(monitorDelivery, "readNativeSupersededMonitorWait").mockImplementation(async (...args) => {
|
||||
if (args[3]) await args[0].execute(sql`set local lock_timeout = '100ms'`);
|
||||
return original(...args);
|
||||
});
|
||||
try {
|
||||
await f.finalize();
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0])
|
||||
.toMatchObject({ phase: "retryable_failure", failureCode: "status_cas_exhausted" });
|
||||
expect(await f.readIssue()).toMatchObject({ status: before.status, statusVersion: before.statusVersion,
|
||||
executionPolicy: before.executionPolicy, monitorNextCheckAt: before.monitorNextCheckAt });
|
||||
expect(await db.select().from(statusDecisions).where(and(eq(statusDecisions.runId, f.runId),
|
||||
eq(statusDecisions.reasonCode, "scheduled_monitor_superseded")))).toEqual([]);
|
||||
} finally { bounded.mockRestore(); release(); await owner; }
|
||||
await f.finalize();
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("committed");
|
||||
expect(await db.select().from(issueComments).where(eq(issueComments.id, comment.id))).toEqual([comment]);
|
||||
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wake.id))).toEqual([wake]);
|
||||
});
|
||||
|
||||
it.each(["comment", "wake", "lineage", "source_stop"])("rechecks newer user-request proof under the task lock (%s)", async state => {
|
||||
const { f, comment, wake, successorId } = await newerUserRequest("consumed");
|
||||
const original = monitorDelivery.readNativeSupersededMonitorWait;
|
||||
const race = vi.spyOn(monitorDelivery, "readNativeSupersededMonitorWait").mockImplementationOnce(async (...args) => {
|
||||
const proof = await original(...args);
|
||||
expect(proof).not.toBeNull();
|
||||
if (state === "comment") await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comment.id));
|
||||
if (state === "wake") await db.update(agentWakeupRequests).set({ status: "cancelled" }).where(eq(agentWakeupRequests.id, wake.id));
|
||||
if (state === "lineage") await db.update(heartbeatRuns).set({ contextSnapshot: {} }).where(eq(heartbeatRuns.id, successorId!));
|
||||
if (state === "source_stop") await db.update(heartbeatRuns).set({ resultJson: sql`jsonb_set(${heartbeatRuns.resultJson}, '{nativeCancellation}', '{"state":"pending"}'::jsonb)` })
|
||||
.where(eq(heartbeatRuns.id, f.runId));
|
||||
return proof;
|
||||
});
|
||||
try { await f.finalize(); } finally { race.mockRestore(); }
|
||||
expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)))[0].phase).toBe("retryable_failure");
|
||||
expect(await db.select().from(statusDecisions).where(and(eq(statusDecisions.runId, f.runId),
|
||||
eq(statusDecisions.reasonCode, "scheduled_monitor_superseded")))).toEqual([]);
|
||||
});
|
||||
|
||||
it.each(["cleared", "owner", "receipt", "rescheduled"])("rechecks monitor proof under the task lock (%s)", async state => {
|
||||
const f = await fixture(), scheduled = await f.readIssue();
|
||||
const [owner] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, status: "running" }).returning();
|
||||
|
||||
@@ -22,7 +22,7 @@ import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.j
|
||||
import { assertNativeCaseHandoffResult } from "./native-case-handoff.js";
|
||||
import { scheduledIssueMonitorWait } from "../issue-monitors.js";
|
||||
import { hasPublishedGitHubReview, readInheritedGitHubReviewWait } from "../chat-channels/github/review-wait.js";
|
||||
import { readNativeMonitorDelivery, readNativePreservedMonitorWait } from "./native-monitor-delivery.js";
|
||||
import { readNativeMonitorDelivery, readNativePreservedMonitorWait, readNativeSupersededMonitorWait } from "./native-monitor-delivery.js";
|
||||
import { isNativePlanWaitResult, readNativePlanWait } from "./native-plan-wait.js";
|
||||
import { hasPendingNativeChildCompletion } from "./native-child-completion-delivery.js";
|
||||
import { settleExternalConversation } from "../slack-conversation-lifecycle.js";
|
||||
@@ -1506,12 +1506,19 @@ export async function finalizeNativeRun(input: {
|
||||
authoritativeIssue.executionRunId !== run.id && authoritativeIssue.checkoutRunId !== run.id
|
||||
? await readNativeMonitorDelivery(input.db, { companyId: run.companyId, issueId: coordinator.issueId, runId: run.id })
|
||||
: null;
|
||||
const supersededMonitorWait = terminalState === "succeeded" && input.workspaceFinalizeStatus === "succeeded" &&
|
||||
assessment.reportedDisposition === "yielded" && assessment.continuation?.kind === "monitor"
|
||||
? await readNativeSupersededMonitorWait(input.db, {
|
||||
companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId,
|
||||
}, authoritativeIssue) : null;
|
||||
if (supersededMonitorWait && supersededMonitorWait.resultId !== resultRow.id) throw new NativeStatusRaceError();
|
||||
const qualityReview = await readRequiredQualityReview(input.db, authoritativeIssue);
|
||||
const statusInput: Parameters<typeof resolveNativeFinalizerStatus>[0] = {
|
||||
requiredQualityReviewSatisfied: qualityReview.satisfied,
|
||||
caseHandoffAuthorized: Boolean(await assertNativeCaseHandoffResult(input.db, run.id, authoritativeIssue.id, result)),
|
||||
monitorWaitDispatched: monitorDelivery !== null,
|
||||
monitorWaitPreserved: preservedMonitorWait !== null,
|
||||
monitorWaitSuperseded: supersededMonitorWait !== null,
|
||||
monitorWaitAuthorized: authoritativeIssue.workMode === "standard" && authoritativeIssue.executionRunId === run.id && monitorWaitAt !== null,
|
||||
planWaitAuthorized: planWait !== null,
|
||||
hasPendingChildCompletion,
|
||||
@@ -1607,6 +1614,7 @@ export async function finalizeNativeRun(input: {
|
||||
monitorDelivery,
|
||||
monitorWaitAt: assessment.continuation?.kind === "monitor" ? monitorWaitAt : null,
|
||||
preservedMonitorWait,
|
||||
supersededMonitorWait,
|
||||
});
|
||||
await input.db
|
||||
.update(nativeRunFinalizations)
|
||||
@@ -1649,6 +1657,8 @@ export async function finalizeNativeRun(input: {
|
||||
? { agentId: run.agentId, nextCheckAt: monitorWaitAt } : undefined,
|
||||
requireMonitorDelivery: decision.reasonCode === "scheduled_monitor_dispatched"
|
||||
? monitorDelivery ?? undefined : undefined,
|
||||
requireSupersededMonitorWait: decision.reasonCode === "scheduled_monitor_superseded"
|
||||
? supersededMonitorWait ?? undefined : undefined,
|
||||
requirePreservedMonitorWait: decision.reasonCode === "scheduled_monitor_preserved"
|
||||
? preservedMonitorWait ?? undefined : undefined,
|
||||
requirePlanWaitSource:
|
||||
|
||||
@@ -150,6 +150,7 @@ export function arbitrateNativeStatus(input: {
|
||||
/** The saved timer already admitted another run and this run no longer owns checkout. */
|
||||
monitorWaitDispatched?: boolean;
|
||||
monitorWaitPreserved?: boolean;
|
||||
monitorWaitSuperseded?: boolean;
|
||||
boardResponseWaitAuthorized?: boolean;
|
||||
boardResponseWaitOrigin?: boolean;
|
||||
isConversation?: boolean;
|
||||
@@ -536,6 +537,11 @@ export function arbitrateNativeStatus(input: {
|
||||
effects: [],
|
||||
};
|
||||
}
|
||||
if (input.monitorWaitSuperseded) return {
|
||||
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
|
||||
statusAction: "preserve", toStatus: input.priorIssueStatus,
|
||||
reasonCode: "scheduled_monitor_superseded", unblockDescriptor: null, effects: [],
|
||||
};
|
||||
if (input.monitorWaitPreserved) {
|
||||
return {
|
||||
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
|
||||
|
||||
@@ -20,7 +20,7 @@ import { hasPublishedGitHubReview, readInheritedGitHubReviewWait } from "../chat
|
||||
import { lockNativeGovernanceSources, pendingNativeGovernance } from "./native-governance.js";
|
||||
import { findSupersededExternalReviewRequests, supersedeExternalReviewRequestsInTransaction } from "./superseded-external-review-requests.js";
|
||||
import { normalizePrpResultSignals, type PrpStructuredRunResult } from "../../vendor/paperclip-runner/index.js";
|
||||
import { readNativeMonitorDelivery, readNativePreservedMonitorWait, type NativeMonitorDelivery, type NativePreservedMonitorWait } from "./native-monitor-delivery.js";
|
||||
import { readNativeMonitorDelivery, readNativePreservedMonitorWait, readNativeSupersededMonitorWait, type NativeSupersededMonitorWait, type NativeMonitorDelivery, type NativePreservedMonitorWait } from "./native-monitor-delivery.js";
|
||||
import { readNativePlanWait, type NativePlanWaitSource } from "./native-plan-wait.js";
|
||||
import { activeIssueInteractionCondition } from "../issue-question-context.js";
|
||||
import { isConversation } from "../agent-conversations.js";
|
||||
@@ -1627,6 +1627,7 @@ export async function commitNativeStatusDecision(input: {
|
||||
requireMonitorWait?: { agentId: string; nextCheckAt: string };
|
||||
requireMonitorDelivery?: NativeMonitorDelivery;
|
||||
requirePreservedMonitorWait?: NativePreservedMonitorWait;
|
||||
requireSupersededMonitorWait?: NativeSupersededMonitorWait;
|
||||
requireNoPendingChildCompletion?: NativeChildCompletionRecipient;
|
||||
requireReassessmentSource?: {
|
||||
phase: string; resultId: string; assessmentId: string; decisionId: string | null; agentId: string;
|
||||
@@ -1850,6 +1851,22 @@ export async function commitNativeStatusDecision(input: {
|
||||
input.decision.statusAction !== "preserve" || input.decision.toStatus !== issue.status ||
|
||||
input.decision.effects.length !== 0) throw new NativeStatusRaceError();
|
||||
}
|
||||
if (reasonCode === "scheduled_monitor_superseded") {
|
||||
const expected = input.requireSupersededMonitorWait;
|
||||
let actual: Awaited<ReturnType<typeof readNativeSupersededMonitorWait>> = null;
|
||||
try {
|
||||
actual = expected ? await readNativeSupersededMonitorWait(tx as unknown as Db, {
|
||||
companyId: input.companyId, issueId: input.issueId, runId: input.runId, agentId: expected.agentId,
|
||||
}, issue, true) : null;
|
||||
} catch (error) {
|
||||
if (record(error).code === "55P03" || record(record(error).cause).code === "55P03") throw new NativeStatusRaceError();
|
||||
throw error;
|
||||
}
|
||||
if (!expected || !actual || actual.resultId !== expected.resultId || actual.wakeId !== expected.wakeId ||
|
||||
actual.commentId !== expected.commentId || actual.successorRunId !== expected.successorRunId ||
|
||||
input.decision.statusAction !== "preserve" || input.decision.toStatus !== issue.status ||
|
||||
input.decision.effects.length !== 0) throw new NativeStatusRaceError();
|
||||
}
|
||||
if (reasonCode === "scheduled_monitor_preserved") {
|
||||
const expected = input.requirePreservedMonitorWait;
|
||||
const actual = expected ? await readNativePreservedMonitorWait(tx as unknown as Db, {
|
||||
|
||||
@@ -3,7 +3,7 @@ import type { Db } from "@paperclipai/db";
|
||||
import { completionContracts, nativeRunResults, workAssessments } from "@paperclipai/db";
|
||||
import { classifyNativeEvidence, type NativeEvidenceAssessment } from "./evidence-classifier.js";
|
||||
import { nativeSha256 } from "./canonical.js";
|
||||
import type { NativeMonitorDelivery, NativePreservedMonitorWait } from "./native-monitor-delivery.js";
|
||||
import type { NativeMonitorDelivery, NativePreservedMonitorWait, NativeSupersededMonitorWait } from "./native-monitor-delivery.js";
|
||||
|
||||
export async function recordNativeWorkAssessment(input: {
|
||||
db: Db;
|
||||
@@ -27,6 +27,7 @@ export async function recordNativeWorkAssessment(input: {
|
||||
/** A persisted schedule can regain settlement authority after a policy repair. */
|
||||
monitorWaitAt?: string | null;
|
||||
preservedMonitorWait?: NativePreservedMonitorWait | null;
|
||||
supersededMonitorWait?: NativeSupersededMonitorWait | null;
|
||||
supersedesAssessmentId?: string | null;
|
||||
}) {
|
||||
const assessmentJson = {
|
||||
@@ -34,6 +35,7 @@ export async function recordNativeWorkAssessment(input: {
|
||||
...(input.monitorDelivery ? { monitorDelivery: input.monitorDelivery } : {}),
|
||||
...(input.monitorWaitAt ? { monitorWaitAt: input.monitorWaitAt } : {}),
|
||||
...(input.preservedMonitorWait ? { preservedMonitorWait: input.preservedMonitorWait } : {}),
|
||||
...(input.supersededMonitorWait ? { supersededMonitorWait: input.supersededMonitorWait } : {}),
|
||||
} as unknown as Record<string, unknown>;
|
||||
const inputDigest = nativeSha256({
|
||||
contractId: input.contractId,
|
||||
|
||||
Reference in new issue
Block a user