mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 10:13:11 +02:00
Merge branch 'codex/dependable-repository-conversion' into codex/dependable-question-journey
* codex/dependable-repository-conversion: fix: defer saved native continuations behind their predecessor
This commit is contained in:
commit
4be3ebe335
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)
|
||||
|
||||
@@ -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"),
|
||||
|
||||
@@ -12,7 +12,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) } },
|
||||
|
||||
@@ -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 &&
|
||||
|
||||
Reference in new issue
Block a user