mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 20:46:28 +01:00
test(runner): add everyday workflow evaluation harness (#13474)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The native runner must support project work, delegation, hiring, and service access. > - Browser tests exposed lost connection access, rejected helper events, and stalled recovery. > - Some eval failures also came from incorrect fixtures and decision controls. > - This pull request fixes those paths and adds eight everyday workflow stories. > - The tests retain observed failures and verify delivered files independently. > - The benefit is repeatable evidence for common user tasks and their remaining gaps. ## Linked Issues or Issue Description Related work: #13404 contains earlier workflow fixes. #13300 and #13470 changed the CI contracts used by the harness security tests. Merged companion: [paperclip-evals#22](https://github.com/paperclipai/paperclip-evals/pull/22). **What happened?** Native ACPX sessions did not receive the assigned connection gateway. Codex helper events could arrive before their spawn receipt and fail thread validation. A parent continuation could take a shared workspace before its child retried. A failed native continuation could leave the task status without a clear recovery blocker. The eval harness also confused tool approvals with new connection requests and could reject a valid delegated download. **Expected behavior** Keep assigned gateway access and its approval checks. Verify helper lineage before accepting helper progress. Let a waiting child proceed before automatic parent recovery. Preserve a failed task's recovery ownership. Grade the actual requested workflow and its delivered files. **Steps to reproduce** Run the everyday workflow suite with the native Codex and Claude profiles. Exercise service approval, connection refusal, delegated project work, and teammate reuse. The commands and case requirements are in `tests/runner-e2e/EVERYDAY-WORKFLOWS.md`. Use `pnpm test:runner-recovery` for controlled crash and replacement cases. ## What Changed - Pass the scoped connection gateway binding through the native ACPX host and sidecar. - Recognize Codex helper lineage from parent metadata and spawn receipts. Verify early helper events with `thread/read`. Keep helper events separate from root completion authority. - Guide agents to use persistent hiring, child tasks, dependency records, and a blocked handoff while waiting for a child. - Defer automatic parent recovery while a child has an active execution path in the same shared workspace. Allow parent recovery when the child needs review. - Record Blocked status and recovery evidence when a failed native continuation needs reconciliation, including existing active or escalated incidents. Preserve their owner and retry budget. - Add eight browser-driven workflow cases. Use real decision controls, explicit child feedback delivery, managed hiring credentials, and independent ZIP checks inside a bounded Docker sandbox. Verify sandbox availability before task creation. Record screenshot SHA-256 at capture. - Keep runner crash probes in controlled recovery tests. Preserve the original failure when cleanup also fails. - Display missing accounting and replay revisions as unavailable. Align harness security assertions with the approved CI changes. - Make the channel-rejection browser fixture bind its file after the send captures its payload. This prevents live refresh from removing the file before the simulated race. ## Verification - Full workspace `pnpm -r typecheck` passed after merging current master. - Runner E2E typecheck passed. Harness unit tests passed: 216/216. - Wake-queue database tests passed: 55/55. The two added existing-incident tests failed before the fix and pass after it. - Docker artifact calibration passed: 12/12. Host-file and host-loopback isolation tests failed before the fix and pass after it. Read-only delivery and output limits are also verified. - Full `pnpm build` passed. Targeted recovery tests passed: 83/83. - The channel-rejection browser test passed five consecutive runs after fixing the fixture race found in CI. - Local general-server (12,351 tests), UI (6,250), CLI (485), and workspace package groups passed. The monolithic run stopped at an unchanged lock-heartbeat fixture race; the isolated workspace group passed on rerun (shared: 747/747). A separate local serialized run passed 97 files before two socket errors in the unchanged issue-list route suite; that suite passed 15/15 on isolated rerun. These local full commands did not finish uninterrupted; the complete CI matrix below covers the remaining suites. - Final head `0fb293733fe307be7e6667ae8f1364077d0c6455`: **34 successful checks, 2 expected skips**, including every server/workspace shard, browser shard, native runner verification, build, and typecheck. [Final CI run](https://github.com/paperclipai/paperclip/actions/runs/34989136700). - Greptile reviewed this exact head at **5/5**; all review threads are resolved. Both Superagent security checks are successful. - ACPX credential-boundary tests passed: 118/118. Superagent accepted the runner/sidecar versus provider-environment trace and cleared its finding. - The latest paid local campaign on source `f6a2fdf7ac2af859826a2ae627ff4125a5478529` passed 22/24 cases: Sol 8/8, Claude 7/8, Mini 7/8. These results predate the merge with current master. - The two remaining failures are in `hire-reuse`: Claude exceeded the attempt deadline during final review; Mini made invalid deliverable tool calls and remained Blocked. - Six Daytona cases were not run because the matching immutable runner image was unavailable. This PR does not claim new remote model results. ## Risks The changes affect connection admission, helper identity, and recovery scheduling. Assigned gateway grants and user approval still govern service calls. The workspace admission gate still exists; the broader folder-sync design is separate work. Provider behavior can still cause the two recorded hiring failures. No database migration is required. Paid cases are opt-in and have bounded attempt deadlines. Project stories now require Docker and the documented pinned Python image on the harness host. ## Model Used OpenAI `gpt-6-astra` performed implementation, diagnosis, and substantive review. OpenAI `gpt-5.6-luna` assisted with verification, PR preparation, and review tracking. Both used repository tools and code execution. Context-window sizes were not recorded. ## 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 (focused checks and isolated reruns; full-run limitations are documented above) - [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: OpenAI GPT-5.6 Luna <noreply@openai.com> Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
58 files changed
+3600
-217
No files matched your search
@@ -12607,6 +12607,44 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
if (retryRun) await waitForRunToSettle(heartbeat, retryRun.id);
|
||||
});
|
||||
|
||||
it("lets a child waiting for the shared workspace run before recovering its lead", async () => {
|
||||
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", runStatus: "succeeded", livenessState: "advanced" });
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId, paperclipWorkspace: { mode: "shared_workspace" } } }).where(eq(heartbeatRuns.id, runId));
|
||||
const projectId = randomUUID(), workspaceId = randomUUID(), childId = randomUUID(), workerId = randomUUID();
|
||||
await db.insert(projects).values({ id: projectId, companyId, name: "Shared project" });
|
||||
await db.insert(projectWorkspaces).values({ id: workspaceId, companyId, projectId, name: "Primary", sourceType: "local_path", cwd: "/tmp/recovery-shared", isPrimary: true });
|
||||
await db.update(issues).set({ projectId, projectWorkspaceId: workspaceId }).where(eq(issues.id, issueId));
|
||||
await db.insert(agents).values({ id: workerId, companyId, name: "Worker", role: "engineer", status: "idle", adapterType: "codex_local", adapterConfig: {} });
|
||||
await db.insert(issues).values({ id: childId, companyId, parentId: issueId, title: "Build the project", status: "in_progress", assigneeAgentId: workerId, projectId, projectWorkspaceId: workspaceId });
|
||||
await db.insert(heartbeatRuns).values({ companyId, agentId: workerId, status: "scheduled_retry", scheduledRetryAt: new Date(Date.now() + 60_000), scheduledRetryReason: "workspace_busy", contextSnapshot: { issueId: childId } });
|
||||
const heartbeat = heartbeatService(db);
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.continuationRequeued).toBe(0);
|
||||
expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId))).toHaveLength(1);
|
||||
// Once the child finishes, normal recovery can continue the lead.
|
||||
await db.update(issues).set({ status: "done" }).where(eq(issues.id, childId));
|
||||
expect((await heartbeat.reconcileStrandedAssignedIssues()).continuationRequeued).toBe(1);
|
||||
const next = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId))).find((r) => r.id !== runId);
|
||||
if (next) await waitForRunToSettle(heartbeat, next.id);
|
||||
});
|
||||
|
||||
it("resumes a shared-workspace lead when its child needs review", async () => {
|
||||
const { companyId, agentId, issueId, runId } = await seedStrandedIssueFixture({ status: "in_progress", runStatus: "succeeded", livenessState: "advanced" });
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: { issueId, paperclipWorkspace: { mode: "shared_workspace" } } }).where(eq(heartbeatRuns.id, runId));
|
||||
const projectId = randomUUID(), workspaceId = randomUUID(), childId = randomUUID(), workerId = randomUUID();
|
||||
await db.insert(projects).values({ id: projectId, companyId, name: "Shared project" });
|
||||
await db.insert(projectWorkspaces).values({ id: workspaceId, companyId, projectId, name: "Primary", sourceType: "local_path", cwd: "/tmp/recovery-shared", isPrimary: true });
|
||||
await db.update(issues).set({ projectId, projectWorkspaceId: workspaceId }).where(eq(issues.id, issueId));
|
||||
await db.insert(agents).values({ id: workerId, companyId, name: "Worker", role: "engineer", status: "idle", adapterType: "codex_local", adapterConfig: {} });
|
||||
await db.insert(issues).values({ id: childId, companyId, parentId: issueId, title: "Review the project", status: "in_review", assigneeAgentId: workerId, projectId, projectWorkspaceId: workspaceId });
|
||||
await db.insert(issueThreadInteractions).values({ companyId, issueId: childId, kind: "request_confirmation", status: "pending", addresseeAgentId: agentId, payload: { prompt: "Review this delivery" } });
|
||||
const heartbeat = heartbeatService(db);
|
||||
const result = await heartbeat.reconcileStrandedAssignedIssues();
|
||||
expect(result.continuationRequeued).toBe(1);
|
||||
const next = (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.agentId, agentId))).find((r) => r.id !== runId);
|
||||
if (next) await waitForRunToSettle(heartbeat, next.id);
|
||||
});
|
||||
|
||||
it("leaves the productive-but-stranded continuation path unchanged under the new classifier", async () => {
|
||||
const { agentId, issueId, runId } = await seedStrandedIssueFixture({
|
||||
status: "in_progress",
|
||||
|
||||
@@ -4,6 +4,7 @@ import { eq } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
activityLog,
|
||||
agentWakeupRequests,
|
||||
agents,
|
||||
companies,
|
||||
@@ -56,6 +57,7 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await db.delete(activityLog);
|
||||
await db.delete(issueComments);
|
||||
// `heartbeat_runs.wakeup_request_id` references `agent_wakeup_requests.id`,
|
||||
// so the run row must go first.
|
||||
@@ -316,6 +318,62 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => {
|
||||
});
|
||||
}
|
||||
|
||||
it.each(["in_progress", "blocked"])("preserves recovery ownership and queued messages when a native task fails from %s", async (status) => {
|
||||
const companyId = await seedCompany();
|
||||
const agentId = await seedAgent({ companyId });
|
||||
const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status });
|
||||
const runId = await seedRun({ companyId, agentId, status: "failed", contextSnapshot: { issueId } });
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", errorCode: "thread_binding_mismatch" }).where(eq(heartbeatRuns.id, runId));
|
||||
const wakeId = await seedDeferredWake({ companyId, agentId, issueId });
|
||||
const adapter = createPostgresWakeQueueAdapter(db, stubDeps);
|
||||
await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => { throw new Error("must not replay an uncertain execution"); });
|
||||
const blockedIssue = (await db.select().from(issues).where(eq(issues.id, issueId)))[0];
|
||||
expect(blockedIssue.status).toBe("blocked");
|
||||
const entries = await db.select().from(activityLog).where(eq(activityLog.entityId, issueId));
|
||||
if (status === "in_progress") {
|
||||
expect(blockedIssue.blockedTransitionAt).not.toBeNull();
|
||||
expect(entries[0]).toMatchObject({ action: "issue.updated", details: { status: "blocked", previousStatus: "in_progress" } });
|
||||
} else expect(entries).toHaveLength(0);
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0].status).toBe("deferred_issue_execution");
|
||||
const action = (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)))[0];
|
||||
expect(action).toMatchObject({ ownerType: "board", cause: "native_continuation_requires_reconciliation" });
|
||||
if (status === "in_progress") expect(action.evidence).toMatchObject({ nativeFailureBlock: { runId, statusVersion: blockedIssue.statusVersion } });
|
||||
else expect(action.evidence.nativeFailureBlock).toBeUndefined();
|
||||
await adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => { throw new Error("must not replay"); });
|
||||
expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId))).toHaveLength(1);
|
||||
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].statusVersion).toBe(blockedIssue.statusVersion);
|
||||
});
|
||||
|
||||
it.each(["active", "escalated"])("repairs a failed native task with an existing %s recovery action", async (status) => {
|
||||
const companyId = await seedCompany();
|
||||
const agentId = await seedAgent({ companyId });
|
||||
const issueId = await seedIssue({ companyId, assigneeAgentId: agentId, status: "in_progress" });
|
||||
const runId = await seedRun({ companyId, agentId, status: "failed", contextSnapshot: { issueId } });
|
||||
await db.update(heartbeatRuns).set({ runtimeMode: "native", errorCode: "runner_lost" }).where(eq(heartbeatRuns.id, runId));
|
||||
const wakeId = await seedDeferredWake({ companyId, agentId, issueId });
|
||||
const [existing] = await db.insert(issueRecoveryActions).values({
|
||||
companyId, sourceIssueId: issueId, status, kind: "active_run_watchdog",
|
||||
ownerType: "board", cause: "native_runner_restart_unverified", fingerprint: `restart:${runId}`,
|
||||
evidence: { runId, priorProof: "keep", automaticRecovery: { attempts: 2 } },
|
||||
nextAction: "Verify the previous execution stopped", attemptCount: 2, maxAttempts: 3,
|
||||
}).returning();
|
||||
const adapter = createPostgresWakeQueueAdapter(db, stubDeps);
|
||||
const release = () => adapter.withIssueExecutionLock({ companyId, runId, now: new Date() }, async () => { throw new Error("must not replay"); });
|
||||
await release();
|
||||
const [blocked] = await db.select().from(issues).where(eq(issues.id, issueId));
|
||||
expect(blocked.status).toBe("blocked");
|
||||
const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId));
|
||||
expect(actions).toHaveLength(1);
|
||||
expect(actions[0]).toMatchObject({ id: existing.id, status, cause: existing.cause,
|
||||
ownerType: "board", attemptCount: 2, maxAttempts: 3, nextAction: existing.nextAction,
|
||||
evidence: { runId, priorProof: "keep", automaticRecovery: { attempts: 2 },
|
||||
nativeFailureBlock: { runId, statusVersion: blocked.statusVersion } } });
|
||||
expect((await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.id, wakeId)))[0].status).toBe("deferred_issue_execution");
|
||||
await release();
|
||||
expect((await db.select().from(issues).where(eq(issues.id, issueId)))[0].statusVersion).toBe(blocked.statusVersion);
|
||||
expect(await db.select().from(activityLog).where(eq(activityLog.entityId, issueId))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each(["queued", "running", "scheduled_retry"])("does not promote another turn behind a %s successor without an execution lock", async (status) => {
|
||||
const companyId = await seedCompany();
|
||||
const agentId = await seedAgent({ companyId });
|
||||
|
||||
@@ -6,6 +6,7 @@ import { and, asc, eq, inArray, isNull, notInArray, or, sql } from "drizzle-orm"
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { extractIssueReferenceIdentifiers } from "@paperclipai/shared";
|
||||
import {
|
||||
activityLog,
|
||||
agentWakeupRequests,
|
||||
agents,
|
||||
chatActions,
|
||||
@@ -720,7 +721,7 @@ async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: HeartbeatRunRow
|
||||
if (!applies || isAcknowledgedNativeStop(run)) return false;
|
||||
|
||||
const existing = await tx
|
||||
.select({ id: issueRecoveryActions.id })
|
||||
.select({ id: issueRecoveryActions.id, evidence: issueRecoveryActions.evidence })
|
||||
.from(issueRecoveryActions)
|
||||
.where(
|
||||
and(
|
||||
@@ -733,6 +734,27 @@ async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: HeartbeatRunRow
|
||||
),
|
||||
)
|
||||
.limit(1);
|
||||
let nativeFailureBlock: { runId: string; statusVersion: number } | undefined;
|
||||
if (issue.status !== "blocked") {
|
||||
const projected = await issueService(tx).update(issue.id, { status: "blocked" }, tx);
|
||||
if (projected) {
|
||||
nativeFailureBlock = { runId: run.id, statusVersion: projected.statusVersion };
|
||||
await tx.insert(activityLog).values({
|
||||
companyId: issue.companyId, actorType: "system", actorId: "execution-recovery",
|
||||
action: "issue.updated", entityType: "issue", entityId: issue.id, runId: run.id,
|
||||
details: { status: "blocked", previousStatus: issue.status, reason: "native_continuation_requires_reconciliation" },
|
||||
});
|
||||
}
|
||||
}
|
||||
// Status projection is required even when restart/finalization created the
|
||||
// incident first. Preserve its owner, cause, retry budget, and prior evidence.
|
||||
if (nativeFailureBlock) {
|
||||
for (const action of existing) {
|
||||
await tx.update(issueRecoveryActions).set({
|
||||
evidence: { ...action.evidence, nativeFailureBlock }, updatedAt: now,
|
||||
}).where(and(eq(issueRecoveryActions.id, action.id), eq(issueRecoveryActions.companyId, issue.companyId)));
|
||||
}
|
||||
}
|
||||
if (!existing.length) {
|
||||
await tx
|
||||
.update(nativeRunFinalizations)
|
||||
@@ -760,7 +782,7 @@ async function recordNativeTerminalRecoveryIfNeeded(tx: Db, run: HeartbeatRunRow
|
||||
returnOwnerAgentId: run.agentId,
|
||||
cause: "native_continuation_requires_reconciliation",
|
||||
fingerprint: `native-continuation:${run.id}`,
|
||||
evidence: { runId: run.id, originalFailureCode: run.errorCode },
|
||||
evidence: { runId: run.id, originalFailureCode: run.errorCode, ...(nativeFailureBlock ? { nativeFailureBlock } : {}) },
|
||||
nextAction:
|
||||
"Inspect the original failure and reconcile the previous execution before continuing. Automatic recovery cannot start another incident.",
|
||||
maxAttempts: 3,
|
||||
|
||||
@@ -15,8 +15,10 @@ import {
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js";
|
||||
import { tmpdir } from "node:os";
|
||||
import { and, eq, inArray } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { join } from "node:path";
|
||||
import { and, eq, inArray, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import {
|
||||
agents,
|
||||
companies,
|
||||
@@ -54,6 +56,11 @@ const support = externalDatabaseUrl
|
||||
);
|
||||
db = createDb(database.connectionString);
|
||||
}, 30_000);
|
||||
afterEach(async () => {
|
||||
// Each sweep scans all companies. Keep earlier tests' unresolved runs out
|
||||
// of later tests so every case exercises only its own recovery fixtures.
|
||||
await db.execute(sql`TRUNCATE companies CASCADE`);
|
||||
});
|
||||
afterAll(async () => {
|
||||
if (externalDatabaseUrl) await db?.$client.end();
|
||||
else await database?.cleanup();
|
||||
@@ -109,6 +116,73 @@ const support = externalDatabaseUrl
|
||||
});
|
||||
return { companyId, agentId, issueId, runId };
|
||||
}
|
||||
it.each(["verified", "unproven", "unknown_action"] as const)(
|
||||
"preserves saved work and a queued request at a controlled recovery boundary (%s)",
|
||||
async (mode) => {
|
||||
// The stopped-session verifier is the injected boundary here. These are
|
||||
// scheduler/database tests, not evidence that SIGKILL is a safe boundary.
|
||||
const workspace = await mkdtemp(join(tmpdir(), "paperclip-recovery-work-"));
|
||||
try {
|
||||
const file = join(workspace, "saved.txt");
|
||||
const saved = "Completed work from before the interruption.\n";
|
||||
await writeFile(file, saved);
|
||||
const source = await seed(2);
|
||||
await db.update(heartbeatRuns).set({ runnerProfileJson: {
|
||||
recoveryEventInventoryVersion: 1,
|
||||
nativeExecutionInput: { provider: { kind: "codex" }, workspace: { cwd: workspace } },
|
||||
} }).where(eq(heartbeatRuns.id, source.runId));
|
||||
await db.update(nativeRunFinalizations).set({ failureCode: "native_session_cleanup_quarantined" })
|
||||
.where(eq(nativeRunFinalizations.runId, source.runId));
|
||||
const projected = await issueService(db).update(source.issueId, { status: "blocked" });
|
||||
await db.insert(issueRecoveryActions).values({
|
||||
companyId: source.companyId, sourceIssueId: source.issueId,
|
||||
kind: "active_run_watchdog", cause: "native_session_cleanup_quarantined", fingerprint: source.runId,
|
||||
ownerType: "board", returnOwnerAgentId: source.agentId, status: "resolved", outcome: "blocked",
|
||||
evidence: { runId: source.runId, nativeFailureBlock: { runId: source.runId, statusVersion: projected!.statusVersion } },
|
||||
nextAction: "Verify the stopped execution before restarting.",
|
||||
});
|
||||
const body = "Keep the saved work and explain the result.";
|
||||
const comment = await issueService(db).addComment(source.issueId, body, { userId: "operator" });
|
||||
if (mode === "unknown_action") await appendHeartbeatRunEvent(db, {
|
||||
companyId: source.companyId, runId: source.runId, agentId: source.agentId,
|
||||
eventType: "tool.execution.started", stream: "system",
|
||||
payload: { name: "send_email", executionId: "unconfirmed-write", transport: "process" },
|
||||
});
|
||||
const retire = vi.fn(() => true);
|
||||
const verifyStoppedSession = vi.fn(async (run: typeof heartbeatRuns.$inferSelect) =>
|
||||
run.id === source.runId && mode !== "unproven" ? { evidence: {}, retire } : null);
|
||||
// Repeated sweeps must not create duplicate successors or user messages.
|
||||
await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession });
|
||||
await reconcileSafeNativeReplacements(db, new Date(), { verifyStoppedSession });
|
||||
const successors = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, source.runId));
|
||||
expect(successors).toHaveLength(mode === "verified" ? 1 : 0);
|
||||
expect(await readFile(file, "utf8")).toBe(saved);
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, source.issueId));
|
||||
expect(comments.filter((row) => row.id === comment.id)).toMatchObject([{ body }]);
|
||||
expect(comments.filter((row) => row.body === body)).toHaveLength(1);
|
||||
const [task] = await db.select().from(issues).where(eq(issues.id, source.issueId));
|
||||
if (mode === "verified") {
|
||||
expect(retire).toHaveBeenCalledOnce();
|
||||
expect(task!.status).toBe("in_progress");
|
||||
const continuation = await buildExecutionContinuation({
|
||||
db, companyId: source.companyId, issueId: source.issueId, agentId: source.agentId,
|
||||
context: successors[0]!.contextSnapshot!, summary: null, exposeLowTrustRaw: false,
|
||||
});
|
||||
expect(continuation.messages.filter((message) => message.id === comment.id)).toHaveLength(1);
|
||||
expect(continuation.objective).toBe(body);
|
||||
} else {
|
||||
expect(retire).not.toHaveBeenCalled();
|
||||
expect(task!.status).toBe("blocked");
|
||||
if (mode === "unknown_action") {
|
||||
const [decision] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, source.runId));
|
||||
expect(decision!.failureDetail?.replacementDenied).toBe("uncertain_provider_action");
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await rm(workspace, { recursive: true, force: true });
|
||||
}
|
||||
},
|
||||
);
|
||||
it.each(["unproven", "changed", "verified"] as const)("requires stopped-session proof through commit (%s)", async (mode) => {
|
||||
const source = await seed(2);
|
||||
await db.update(nativeRunFinalizations).set({ failureCode: "native_session_cleanup_quarantined" }).where(eq(nativeRunFinalizations.runId, source.runId));
|
||||
|
||||
@@ -2829,7 +2829,8 @@ export function recoveryService(
|
||||
);
|
||||
}
|
||||
|
||||
async function healthyOpenChildIssues(issue: typeof issues.$inferSelect) {
|
||||
async function healthyOpenChildIssues(issue: typeof issues.$inferSelect, sameWorkspaceOnly = false) {
|
||||
if (sameWorkspaceOnly && !issue.projectWorkspaceId) return [];
|
||||
const childCandidates = await db
|
||||
.select()
|
||||
.from(issues)
|
||||
@@ -2837,6 +2838,7 @@ export function recoveryService(
|
||||
and(
|
||||
eq(issues.companyId, issue.companyId),
|
||||
eq(issues.parentId, issue.id),
|
||||
...(sameWorkspaceOnly ? [eq(issues.projectWorkspaceId, issue.projectWorkspaceId!)] : []),
|
||||
visibleIssueCondition(),
|
||||
notInArray(issues.status, ["done", "cancelled"]),
|
||||
),
|
||||
@@ -2848,7 +2850,7 @@ export function recoveryService(
|
||||
});
|
||||
if (
|
||||
childState.hasActiveExecutionPath ||
|
||||
childState.hasDurableWaitingPath
|
||||
(!sameWorkspaceOnly && childState.hasDurableWaitingPath)
|
||||
) {
|
||||
openChildren.push({ id: child.id, identifier: child.identifier });
|
||||
}
|
||||
@@ -4996,6 +4998,17 @@ export function recoveryService(
|
||||
if (isSuccessfulInProgressContinuationRun(latestRun)) {
|
||||
const successfulRun = latestRun;
|
||||
|
||||
// A child with a live or durable waiting path must get a chance to use
|
||||
// the shared workspace. Repeated automatic parent continuations can
|
||||
// otherwise reacquire it before the child's resource retry is due.
|
||||
// This only gates recovery; explicit messages still follow admission.
|
||||
const workspace = parseObject(parseObject(successfulRun.contextSnapshot).paperclipWorkspace);
|
||||
if (workspace.mode === "shared_workspace" && (await healthyOpenChildIssues(issue, true)).length > 0) {
|
||||
result.productiveContinuationObserved += 1;
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!isProductiveContinuationRun(successfulRun)) {
|
||||
result.successfulContinuationObserved += 1;
|
||||
result.skipped += 1;
|
||||
|
||||
Reference in new issue
Block a user