Merge branch 'codex/dependable-question-journey' into codex/dependable-qualification-checkpoint

* codex/dependable-question-journey:
  fix: defer saved native continuations behind their predecessor
This commit is contained in:
Dotta committed 2026-10-21 10:55:48 -05:00
commit 0819b36f0c
8 files changed
+118 -8

No files matched your search

@@ -8,6 +8,12 @@ import {
agentRuntimeState,
agentWakeupRequests,
companySkills,
completionContracts,
nativeRunResults,
nativeRunFinalizations,
workAssessments,
statusDecisions,
statusDecisionEffects,
companies,
authUsers,
companyMemberships,
@@ -144,6 +150,12 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () =
await db.delete(documents);
await db.delete(issueRelations);
await db.delete(issueTreeHolds);
await db.delete(statusDecisionEffects);
await db.delete(statusDecisions);
await db.delete(workAssessments);
await db.delete(nativeRunFinalizations);
await db.delete(nativeRunResults);
await db.delete(completionContracts);
await db.delete(issues);
await db.delete(heartbeatRunEvents);
await db.delete(activityLog);
@@ -259,6 +271,76 @@ describeEmbeddedPostgres("heartbeat dependency-aware queued run selection", () =
}
}, 45_000);
it("keeps a saved native continuation behind its active predecessor without retagging that run", async () => {
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
let finishPredecessor!: () => void;
const predecessorFinished = new Promise<void>(resolve => { finishPredecessor = resolve; });
mockAdapterExecute.mockImplementationOnce(async () => {
await predecessorFinished;
return { exitCode: 0, signal: null, timedOut: false, errorMessage: null,
summary: "The repository is staged; continue on the saved admission.", provider: "test", model: "test-model" };
});
mockAdapterExecute.mockImplementationOnce(async () => {
await db.update(issues).set({ status: "done" }).where(eq(issues.id, issueId));
return { exitCode: 0, signal: null, timedOut: false, errorMessage: null,
summary: "Verified and delivered the prepared repository.", provider: "test", model: "test-model" };
});
await db.insert(companies).values({ id: companyId, name: "Saved repository continuation",
issuePrefix: `N${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
requireBoardApprovalForNewAgents: false, defaultResponsibleUserId: "responsible-user" });
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "responsible-user", status: "active", membershipRole: "member" });
await db.insert(agents).values({ id: agentId, companyId, name: "Repository worker", role: "engineer", status: "active",
adapterType: "codex_local", adapterConfig: {}, permissions: {}, runtimeConfig: { heartbeat: { wakeOnDemand: true, maxConcurrentRuns: 1 } } });
await db.insert(issues).values({ id: issueId, companyId, title: "Continue repository verification", status: "todo",
assigneeAgentId: agentId, responsibleUserId: "responsible-user" });
try {
const predecessor = await heartbeat.wakeup(agentId, { source: "assignment", triggerDetail: "system", reason: "issue_assigned",
payload: { issueId }, contextSnapshot: { issueId, source: "initial-repository-request" } });
expect(predecessor).not.toBeNull();
expect(await waitForCondition(async () => mockAdapterExecute.mock.calls.length === 1, 30_000)).toBe(true);
const contractId = randomUUID(), resultId = randomUUID(), assessmentId = randomUUID(), decisionId = randomUUID();
await db.insert(completionContracts).values({ id: contractId, companyId, issueId, revision: 1,
schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard",
completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
contractJson: {}, canonicalSha256: contractId, createdByActorType: "system", createdByActorId: "test" });
await db.update(heartbeatRuns).set({ nativeIssueId: issueId, completionContractId: contractId }).where(eq(heartbeatRuns.id, predecessor!.id));
await db.insert(nativeRunResults).values({ id: resultId, companyId, issueId, runId: predecessor!.id,
completionContractId: contractId, serverFingerprint: resultId, schemaStatus: "accepted", resultJson: {}, canonicalSha256: resultId });
await db.insert(workAssessments).values({ id: assessmentId, companyId, issueId, runId: predecessor!.id,
contractId, resultId, triggerKind: "turn_finished", triggerActorCompanyId: companyId,
priorIssueStatus: "in_progress", priorStatusVersion: 0, policyVersion: "test", assessmentJson: {}, inputDigest: assessmentId });
await db.insert(statusDecisions).values({ id: decisionId, companyId, issueId, runId: predecessor!.id,
assessmentId, decisionVersion: 1, policyVersion: "test", fromStatus: "in_progress", toStatus: "in_progress",
reasonCode: "live_continuation_registered", decisionJson: {}, decisionDigest: decisionId, applicationState: "applied", appliedAt: new Date() });
const [intent] = await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", triggerDetail: "system",
reason: "issue_status_changed", requestedByActorType: "system", requestedByActorId: "native-status-committer",
idempotencyKey: `native-status:${decisionId}:continuation`, payload: { issueId, taskId: issueId, nativeDecisionId: decisionId, continuationKind: "same_agent",
_paperclipWakeContext: { issueId, taskId: issueId, source: "native_status_decision", continuationKind: "same_agent" } } }).returning();
await db.insert(statusDecisionEffects).values({ companyId, issueId, decisionId, ordinal: 0,
effectKind: "enqueue_continuation", targetType: "agent_wakeup_request", targetId: intent!.id,
idempotencyKey: `native-status:${decisionId}:continuation`, payload: { continuationKind: "same_agent" }, deliveryState: "delivered", deliveredAt: new Date() });
await heartbeat.dispatchPendingNativeStatusWakeups({ companyId });
await heartbeat.dispatchPendingNativeStatusWakeups({ companyId });
const queued = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId));
expect(queued.filter(row => row.status === "deferred_issue_execution"), JSON.stringify(queued.map(row => ({ status: row.status, reason: row.reason, runId: row.runId, error: row.error })))).toHaveLength(1);
expect(queued.find(row => row.id === intent!.id)).toMatchObject({ status: "coalesced", runId: null });
const [unchanged] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, predecessor!.id));
expect(unchanged!.contextSnapshot).toMatchObject({ source: "initial-repository-request" });
expect(unchanged!.contextSnapshot?.nativeStatusWakeIntentId).toBeUndefined();
expect(mockAdapterExecute).toHaveBeenCalledTimes(1);
finishPredecessor();
expect(await waitForCondition(async () => mockAdapterExecute.mock.calls.length === 2, 10_000)).toBe(true);
await heartbeat.drainActiveRunExecutions();
const completed = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId));
expect(completed).toHaveLength(2);
const successor = completed.find(run => run.id !== predecessor!.id)!;
expect(successor).toMatchObject({ status: "succeeded", contextSnapshot: { source: "native_status_decision", nativeStatusWakeIntentId: intent!.id } });
expect(successor.startedAt!.getTime()).toBeGreaterThanOrEqual(completed.find(run => run.id === predecessor!.id)!.finishedAt!.getTime());
await heartbeat.dispatchPendingNativeStatusWakeups({ companyId });
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId))).toHaveLength(2);
} finally { finishPredecessor(); }
}, 45_000);
it("dispatches and coalesces durable native status wake intents into one heartbeat run", async () => {
const companyId = randomUUID();
const agentId = randomUUID();
@@ -196,6 +196,15 @@ if (!support.supported) console.warn(`Monitor publication database unavailable:
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.source.id))).toHaveLength(1);
});
it("retains conversation provenance when the admitted dispatch owns the deferred outbox run binding", async () => {
const f = await continuedTaskFixture();
await db.update(agentWakeupRequests).set({ runId: null }).where(eq(agentWakeupRequests.id, f.continuationWake.id));
expect(await f.bindings()).toEqual(await f.bindings(f.source.id));
expect(await f.bindings()).toHaveLength(1);
await db.update(agentWakeupRequests).set({ runId: f.source.id }).where(eq(agentWakeupRequests.id, f.continuationWake.id));
expect(await f.bindings()).toEqual([]);
});
it("selects the continued chat run's completed answer over an earlier bookkeeping comment", async () => {
const f = await continuedTaskFixture();
const authorization = isChatOriginPublicationCandidate(f.successor.contextSnapshot)
+1
View File
@@ -4778,6 +4778,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende
try {
const wakeRun = await enqueueWakeup(candidate.agentId, {
allowRunCoalescing: false,
source: candidate.source as WakeupOptions["source"],
triggerDetail: (candidate.triggerDetail ??
"system") as WakeupOptions["triggerDetail"],
@@ -1,5 +1,5 @@
import { agentWakeupRequests, heartbeatRuns, statusDecisions, statusDecisionEffects, type Db } from "@paperclipai/db";
import { and, eq, inArray, ne, sql } from "drizzle-orm";
import { and, eq, inArray, isNull, ne, or, sql } from "drizzle-orm";
import { alias } from "drizzle-orm/pg-core";
import { boundRunIssueIdSql } from "../run-task-binding.js";
@@ -14,7 +14,7 @@ export async function readNativeContinuationSource(
const sources = await db.select({ runId: source.id }).from(heartbeatRuns)
.innerJoin(agentWakeupRequests, and(
eq(agentWakeupRequests.companyId, binding.companyId),
eq(agentWakeupRequests.runId, heartbeatRuns.id),
or(isNull(agentWakeupRequests.runId), eq(agentWakeupRequests.runId, heartbeatRuns.id)),
eq(agentWakeupRequests.agentId, heartbeatRuns.agentId),
eq(agentWakeupRequests.reason, "issue_status_changed"),
eq(agentWakeupRequests.status, "coalesced"),
+3 -1
View File
@@ -46,7 +46,9 @@ observer must download that source-attributed artifact while only the first
admission is running, before the successor starts. A missing original, late
observation, or different successor-created reference fails qualification. The
successor is instructed to stop on a missing file and never recreate it. Both admissions
must retain one workspace and commit their native results. The fixture uses
must retain one workspace and commit their native results. The successor must
have its own dispatch receipt and saved native intent. It starts after the
predecessor finishes; dispatch must not retag the predecessor. The fixture uses
production hire instructions and records the selected catalog model. These four
development cells do not qualify Slack, staging configurations, migration,
finalization restart, or the complete dependable-work milestone.
@@ -27,7 +27,7 @@ export async function runAutomaticRepositoryFlow(input: {
const observation: AutomaticRepositoryObservation = { reference: "", issueDone: false, runCount: 0,
successfulCommittedRuns: 0, sourceContinuation: null, successorSource: null, workspaceIds: [], readyRepository: false,
savedContext: null, firstSavedContext: null, firstContextAttributed: false,
firstContextObservedAt: null, successorStartedAt: null,
firstContextObservedAt: null, successorStartedAt: null, predecessorFinishedAt: null, predecessorRetagged: true, continuationReceipt: null,
actualReadme: null, actualCommit: null, downloaded: null, downloadVerified: false,
visibleDelivery: false, pendingInteractions: -1, monitorScheduled: false };
async function refresh() {
@@ -44,6 +44,10 @@ export async function runAutomaticRepositoryFlow(input: {
observation.sourceContinuation = runs[0]?.resultJson?.nativeResult?.continuation?.kind ?? null;
observation.successorSource = runs[1]?.contextSnapshot?.source ?? null;
observation.successorStartedAt = runs[1]?.startedAt ? Date.parse(runs[1].startedAt) : null;
observation.predecessorFinishedAt = runs[0]?.finishedAt ? Date.parse(runs[0].finishedAt) : null;
observation.predecessorRetagged = Boolean(runs[0]?.contextSnapshot?.nativeStatusWakeIntentId);
const intentId = runs[1]?.contextSnapshot?.nativeStatusWakeIntentId, dispatchId = runs[1]?.wakeupRequestId;
observation.continuationReceipt = typeof intentId === "string" && typeof dispatchId === "string" ? { intentId, dispatchId } : null;
if (runs.length === 1 && runs[0]?.status === "running" && observation.firstContextObservedAt === null) {
const original = await readRegisteredArtifacts(api, issue.id, runs[0].id, [files.context]).catch(() => null);
if (original?.[0]) {
@@ -6,7 +6,8 @@ const reference = "c".repeat(64);
const valid: AutomaticRepositoryObservation = { reference, issueDone: true, runCount: 2,
successfulCommittedRuns: 2, sourceContinuation: "same_agent", successorSource: "native_status_decision", workspaceIds: ["workspace", "workspace"],
readyRepository: true, savedContext: `${reference}\n`, firstSavedContext: `${reference}\n`, firstContextAttributed: true,
firstContextObservedAt: 1000, successorStartedAt: 2000,
firstContextObservedAt: 1000, successorStartedAt: 2000, predecessorFinishedAt: 1900, predecessorRetagged: false,
continuationReceipt: { intentId: "11111111-1111-4111-8111-111111111111", dispatchId: "22222222-2222-4222-8222-222222222222" },
actualReadme: "Hello World!\n", actualCommit: sha,
downloaded: { reference, readme: "Hello World!\n", commit: sha }, downloadVerified: true,
visibleDelivery: true, pendingInteractions: 0, monitorScheduled: false };
@@ -36,7 +37,10 @@ it.each<Partial<AutomaticRepositoryObservation>>([
{ workspaceIds: ["old", "new"] }, { savedContext: null }, { firstSavedContext: null }, { firstContextAttributed: false },
{ firstSavedContext: `${"d".repeat(64)}\n` }, { firstSavedContext: "REFERENCE-fresh\n" },
{ firstContextObservedAt: null }, { firstContextObservedAt: 2000 }, { firstContextObservedAt: 3000 }, { successorStartedAt: null },
{ readyRepository: false }, { downloaded: null },
{ readyRepository: false }, { downloaded: null }, { predecessorFinishedAt: null }, { predecessorFinishedAt: 2001 },
{ predecessorRetagged: true }, { continuationReceipt: null },
{ continuationReceipt: { ...valid.continuationReceipt!, dispatchId: valid.continuationReceipt!.intentId } },
{ continuationReceipt: { ...valid.continuationReceipt!, intentId: "unverified" } },
{ downloaded: { ...valid.downloaded, reference: "REFERENCE-stale" } },
{ downloaded: { ...valid.downloaded, readme: "I inspected it and all checks pass" } },
{ downloaded: { ...valid.downloaded, commit: "b".repeat(40) } },
+10 -2
View File
@@ -42,6 +42,9 @@ export interface AutomaticRepositoryObservation {
firstContextAttributed: boolean;
firstContextObservedAt: number | null;
successorStartedAt: number | null;
predecessorFinishedAt: number | null;
predecessorRetagged: boolean;
continuationReceipt: { intentId: string; dispatchId: string } | null;
actualReadme: string | null;
actualCommit: string | null;
downloaded: { reference?: unknown; readme?: unknown; commit?: unknown } | null;
@@ -65,8 +68,13 @@ export function gradeAutomaticRepository(value: AutomaticRepositoryObservation)
return [
check("durable-completion", value.issueDone && value.runCount === 2 && value.successfulCommittedRuns === 2,
"Two committed provider admissions must finish the requested result."),
check("server-owned-continuation", value.sourceContinuation === "same_agent" && value.successorSource === "native_status_decision" && !value.monitorScheduled,
"Repository setup must continue through the saved same-agent wake without a timer or user message."),
check("server-owned-continuation", value.sourceContinuation === "same_agent" && value.successorSource === "native_status_decision" && !value.monitorScheduled &&
!value.predecessorRetagged && value.predecessorFinishedAt !== null && value.successorStartedAt !== null &&
value.successorStartedAt >= value.predecessorFinishedAt && value.continuationReceipt !== null &&
/^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i.test(value.continuationReceipt.intentId) &&
/^[a-f0-9]{8}(?:-[a-f0-9]{4}){3}-[a-f0-9]{12}$/i.test(value.continuationReceipt.dispatchId) &&
value.continuationReceipt.intentId !== value.continuationReceipt.dispatchId,
"The saved same-agent intent must bind a distinct successor dispatch after the unmodified predecessor finishes, without a timer or user message."),
check("retained-workspace", value.workspaceIds.length === 2 && Boolean(value.workspaceIds[0]) && value.workspaceIds.every(id => id === value.workspaceIds[0]) && value.readyRepository,
"Both admissions must retain the same task workspace and its prepared repository."),
check("retained-context", value.firstContextAttributed && value.firstContextObservedAt !== null && value.successorStartedAt !== null &&