mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-25 09:41:04 +01:00
fix(recovery): project context evidence for latest-run checks (#17463)
## Thinking Path > - Paperclip manages AI agents and their tasks. > - Recovery sweeps decide whether tasks have a valid next step. > - Those decisions inspect saved run metadata. > - Latest-run queries also return full prompts and execution setup data. > - This repeats large responses when a sweep leaves a task unchanged. > - This change projects the context evidence recovery actually reads. > - A row-version check keeps stale evidence from changing a task. ## Linked Issues or Issue Description **What happened?** The latest-run recovery queries retrieve large stored contexts for each candidate. The decision usually consumes small recovery, message, or workspace-mode fields. **Expected behavior** Keep all decision evidence while selecting a small response. Preserve complete stored contexts and full execution reads. Detect concurrent source changes. **Steps to reproduce** 1. Store a terminal run with large task Markdown, wake, workspace, and continuation bodies. 2. Call `reconcileStrandedAssignedIssues` for its assigned task. 3. Observe the full context returned by the latest-run query on each sweep. **Paperclip version or commit** Baseline: `410e5c6eedba98588a87e4104153898fb0437532`. Related read optimization: Refs #17405. Complementary fix: Refs #17460. These PRs have no code dependency. Searched existing PRs and issues for run projections, recovery reads, and database transfer. No duplicate was found. **Deployment mode** External PostgreSQL. Verification uses disposable real PostgreSQL and synthetic data. ## What Changed - Project typed context evidence in all three latest-run recovery queries. - Keep only the used workspace mode and external-interaction flag from nested setup objects. - Preserve original JSON types, message references, retry evidence, and existing query scopes and order. - Read PostgreSQL `xmin` with the run. Require the same row version during native blocked-wait repair. - Retain the original full-context comparison for callers that supply a full run without a row version. - Test large payloads, malformed JSON shapes, review-message recovery, and concurrent source edits. ## Verification - Final-source integration checks: **450 passed** across context projections, native finalization recovery, and the assembled heartbeat recovery service. - Focused issue-lock race check: **9 passed** (401 outside the filter), including omitted-context and late-output edits and successful repair on a fresh sweep. - Final-source recursive typecheck and full build pass. The full local stable suite is still running; current-head CI must also pass before merge. - Real-service SQL capture against the baseline: one returned row falls from 699,724 to 573 serialized JSON bytes (99.918% smaller). - The fixture uses four 174,764-byte incompressible execution fields. All unaffected returned columns compare equal. Regression tests check the retained decision evidence and unchanged stored records. - Five alternating warm samples after two warmups: baseline median 1.479 ms, treatment 1.084 ms. Ranges overlap; no request-latency speedup is claimed. Plans were captured outside the timing interval. - An initial full-context hash projection cut bytes but raised the local median to 7.457 ms. The final row-version check removes that cost. - One integration run overlapped the hash-export rename and failed seven tests on the removed export. A fresh final-source run passed all 450 tests. The earlier mixed-source failure remains recorded and is not counted as a pass. - All current-head CI checks pass on `17add20e391c30ed0496f4b39eb9c59779efa6b4`. Greptile: **5/5, medium impact**, no findings or unresolved threads on that head. - Hosted transfer and product behavior remain to be verified on staging after merge. ## Risks The context selection must track its recovery consumers. Tests preserve typed flags, retry budgets, external-chat references, and malformed-input behavior. The row-version guard is conservative: any intervening run update defers native blocker repair to a fresh sweep. Concurrent omitted-context and output edits are covered through the real service and issue-lock wait; a fresh sweep must still repair. There is no migration, prompt truncation, history deletion, or authorization cache. ## Model Used OpenAI GPT-6 through Codex, with source inspection, code execution, and GitHub tools. The exact served model ID and context-window size are not exposed in this session. ## 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:
1 parent
60ca3a903a
commit
7bd449dbb6
5 files changed
+147
-6
No files matched your search
@@ -49,6 +49,13 @@ and leaves all existing cases passive. Source identities are not grants.
|
||||
|
||||
### Task query performance
|
||||
|
||||
Recovery's latest-run queries return the context fields used by recovery
|
||||
decisions. Prompts, execution continuation bodies, and workspace setup payloads
|
||||
remain in the stored run and its full detail reads. These queries also return
|
||||
the PostgreSQL row version. Native blocked-wait repair compares that version
|
||||
against a fresh full source read before applying changes. Any intervening row
|
||||
update, including an edit outside the projection, defers repair to a fresh sweep.
|
||||
|
||||
Run-context lookups use expression indexes for both `context_snapshot.issueId`
|
||||
and `context_snapshot.paperclipIssue.id`. Task creation provenance has its own
|
||||
company-scoped expression index that follows the query's native issue ID and
|
||||
|
||||
@@ -15676,6 +15676,8 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
"comment",
|
||||
"wake",
|
||||
"run",
|
||||
"run_context",
|
||||
"run_metadata",
|
||||
"attention",
|
||||
] as const)(
|
||||
"does not clobber a newer %s while native blocker repair waits on the issue lock",
|
||||
@@ -15750,6 +15752,12 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
continuationPolicy: "wake_assignee",
|
||||
payload: { version: 1, questions: [] },
|
||||
});
|
||||
if (change === "run_context")
|
||||
await tx.update(heartbeatRuns).set({
|
||||
contextSnapshot: sql`jsonb_set(${heartbeatRuns.contextSnapshot}, '{paperclipTaskMarkdown}', '"Changed execution instructions"'::jsonb)`,
|
||||
}).where(eq(heartbeatRuns.id, runId));
|
||||
if (change === "run_metadata")
|
||||
await tx.update(heartbeatRuns).set({ stdoutExcerpt: "Late execution output" }).where(eq(heartbeatRuns.id, runId));
|
||||
[snapshot] = await tx
|
||||
.select()
|
||||
.from(issues)
|
||||
@@ -15793,6 +15801,10 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => {
|
||||
),
|
||||
).toHaveLength(0);
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
if (change === "run_context" || change === "run_metadata") {
|
||||
expect((await heartbeat.reconcileStrandedAssignedIssues()).escalated).toBe(1);
|
||||
expect(mockAdapterExecute).not.toHaveBeenCalled();
|
||||
}
|
||||
},
|
||||
);
|
||||
|
||||
|
||||
@@ -0,0 +1,84 @@
|
||||
import { randomBytes, randomUUID } from "node:crypto";
|
||||
import { agents, companies, createDb, heartbeatRuns } from "@paperclipai/db";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
|
||||
import { isExternalChatPresentationContext } from "../heartbeat-run-summary.js";
|
||||
import { legacyDispositionEpisode } from "./legacy-continuation.js";
|
||||
import { decideIssueReviewPathRecovery } from "./review-path-recovery.js";
|
||||
import { recoveryRunContextColumn, recoveryRunVersionColumn } from "./run-context-read.js";
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
|
||||
describe.skipIf(!support.supported)("recovery context PostgreSQL projection", () => {
|
||||
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let companyId: string;
|
||||
let agentId: string;
|
||||
beforeAll(async () => {
|
||||
database = await startEmbeddedPostgresTestDatabase("recovery-context-read-");
|
||||
db = createDb(database.connectionString);
|
||||
const [company] = await db.insert(companies).values({ name: "Recovery fixture", issuePrefix: "RC" }).returning();
|
||||
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Recovery agent", adapterType: "process" }).returning();
|
||||
companyId = company.id;
|
||||
agentId = agent.id;
|
||||
}, 60_000);
|
||||
afterAll(async () => {
|
||||
await db?.$client.end();
|
||||
await database?.cleanup();
|
||||
});
|
||||
|
||||
async function read(contextSnapshot: unknown) {
|
||||
const [run] = await db.insert(heartbeatRuns).values({ companyId, agentId, contextSnapshot: contextSnapshot as Record<string, unknown> | null }).returning();
|
||||
const [selected] = await db.select({ context: recoveryRunContextColumn, version: recoveryRunVersionColumn })
|
||||
.from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
return { run, selected };
|
||||
}
|
||||
|
||||
it("retains typed recovery evidence while excluding large execution data", async () => {
|
||||
const issueId = randomUUID();
|
||||
const payload = randomBytes(128 * 1024).toString("base64");
|
||||
const evidence = {
|
||||
issueId, taskId: issueId, retryReason: "issue_continuation_needed", source: "chat:slack",
|
||||
wakeReason: "assignment", handoffReason: "successful_run_missing_state", handoffRequired: true,
|
||||
handoffAttempt: 2, maxHandoffAttempts: 3, sourceRunId: randomUUID(), resumeFromRunId: randomUUID(),
|
||||
missingDisposition: "clear_next_step", providerQuotaRetryNotBefore: "2099-01-01T00:00:00Z",
|
||||
transientRetryNotBefore: "2099-01-02T00:00:00Z", dispositionRepairAttempt: 2,
|
||||
dispositionRepairFingerprint: "fingerprint", legacyDispositionEpisode: { id: "episode", attempt: 1, maxAttempts: 2 },
|
||||
reviewPathRecoveryAttempt: 1, reviewPathConsumedRef: "reference", interactionId: randomUUID(),
|
||||
approvalId: randomUUID(), clearReason: "resolved", monitorAttemptCount: 4,
|
||||
wakeCommentIds: [randomUUID(), randomUUID()], wakeCommentId: randomUUID(), commentId: randomUUID(),
|
||||
externalChatContinuation: true,
|
||||
};
|
||||
const context = { ...evidence, paperclipWake: { externalInteractionContinuation: true, prompt: payload },
|
||||
paperclipWorkspace: { mode: "shared_workspace", instructions: payload }, executionContinuation: { prompt: payload }, paperclipTaskMarkdown: payload };
|
||||
const { run, selected } = await read(context);
|
||||
expect(selected.context).toEqual({ ...evidence, paperclipWake: { externalInteractionContinuation: true }, paperclipWorkspace: { mode: "shared_workspace" } });
|
||||
expect(Buffer.byteLength(JSON.stringify(selected))).toBeLessThan(4096);
|
||||
expect(isExternalChatPresentationContext(selected.context)).toBe(isExternalChatPresentationContext(context));
|
||||
expect(legacyDispositionEpisode({ ...run, contextSnapshot: selected.context })).toEqual(legacyDispositionEpisode(run));
|
||||
const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(stored.contextSnapshot).toEqual(context);
|
||||
const [serialized] = await db.$client.unsafe("SELECT xmin::text AS version FROM heartbeat_runs WHERE id = $1", [run.id]);
|
||||
expect(selected.version).toBe(serialized.version);
|
||||
await db.update(heartbeatRuns).set({ contextSnapshot: { ...context, paperclipTaskMarkdown: "changed" } }).where(eq(heartbeatRuns.id, run.id));
|
||||
const [changed] = await db.select({ context: recoveryRunContextColumn, version: recoveryRunVersionColumn }).from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
||||
expect(changed.context).toEqual(selected.context);
|
||||
expect(changed.version).not.toBe(selected.version);
|
||||
});
|
||||
|
||||
it.each([null, [], ["unexpected"], "scalar", 42, { issueId: 42, taskId: false, paperclipWake: [], paperclipWorkspace: null }].map(context => ({ context })))("preserves decisions for malformed context $context", async ({ context }) => {
|
||||
const { selected } = await read(context);
|
||||
expect(isExternalChatPresentationContext(selected.context)).toBe(isExternalChatPresentationContext(context));
|
||||
expect(legacyDispositionEpisode({ id: "run", contextSnapshot: selected.context })).toEqual(legacyDispositionEpisode({ id: "run", contextSnapshot: context }));
|
||||
});
|
||||
|
||||
it("retains external-chat review recovery message references and monitor identity", async () => {
|
||||
const context = { source: "chat:slack", wakeReason: "issue_monitor_fired", clearReason: "fired", monitorAttemptCount: 3,
|
||||
wakeCommentIds: [randomUUID(), randomUUID()], prompt: randomBytes(64 * 1024).toString("base64") };
|
||||
const { selected } = await read(context);
|
||||
const input = { issueId: randomUUID(), sourceRunId: randomUUID(), assigneeAgentId: agentId, reviewAttention: { state: "stalled" } as Parameters<typeof decideIssueReviewPathRecovery>[0]["reviewAttention"], existingWake: false };
|
||||
expect(decideIssueReviewPathRecovery({ ...input, contextSnapshot: selected.context }))
|
||||
.toEqual(decideIssueReviewPathRecovery({ ...input, contextSnapshot: context }));
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,28 @@
|
||||
import { heartbeatRuns } from "@paperclipai/db";
|
||||
import { sql } from "drizzle-orm";
|
||||
|
||||
const recoveryContextKeys = [
|
||||
"issueId", "taskId", "retryReason", "source", "wakeReason",
|
||||
"handoffReason", "handoffRequired", "handoffAttempt", "maxHandoffAttempts",
|
||||
"sourceRunId", "resumeFromRunId", "missingDisposition",
|
||||
"providerQuotaRetryNotBefore", "transientRetryNotBefore",
|
||||
"dispositionRepairAttempt", "dispositionRepairFingerprint", "legacyDispositionEpisode",
|
||||
"reviewPathRecoveryAttempt", "reviewPathConsumedRef", "interactionId", "approvalId",
|
||||
"clearReason", "monitorAttemptCount", "wakeCommentIds", "wakeCommentId", "commentId",
|
||||
"externalChatContinuation", "paperclipWake", "paperclipWorkspace",
|
||||
] as const;
|
||||
|
||||
export const recoveryRunContextColumn = sql<Record<string, unknown> | null>`
|
||||
case when ${heartbeatRuns.contextSnapshot} is null then null else (
|
||||
select coalesce(jsonb_object_agg(field.key, case field.key
|
||||
when 'paperclipWake' then jsonb_build_object(
|
||||
'externalInteractionContinuation', field.value->'externalInteractionContinuation')
|
||||
when 'paperclipWorkspace' then jsonb_build_object('mode', field.value->'mode')
|
||||
else field.value end), '{}'::jsonb)
|
||||
from jsonb_each(case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object'
|
||||
then ${heartbeatRuns.contextSnapshot} else '{}'::jsonb end) as field(key, value)
|
||||
where field.key in (${sql.join(recoveryContextKeys.map(key => sql`${key}`), sql`, `)})
|
||||
) end
|
||||
`;
|
||||
|
||||
export const recoveryRunVersionColumn = sql<string>`${heartbeatRuns}.xmin::text`;
|
||||
@@ -1,3 +1,4 @@
|
||||
import { recoveryRunContextColumn, recoveryRunVersionColumn } from "./run-context-read.js";
|
||||
import { isAgentAwaitingSetup } from "../../modules/agent-lifecycle/index.js";
|
||||
import { onboardingFirstTaskAwaitingUserSql } from "./onboarding-first-task-wait.js";
|
||||
import { isAiAuthenticationBlocked } from "../ai-auth-failure.js";
|
||||
@@ -25,6 +26,7 @@ import {
|
||||
desc,
|
||||
eq,
|
||||
gt,
|
||||
getTableColumns,
|
||||
gte,
|
||||
inArray,
|
||||
isNotNull,
|
||||
@@ -247,6 +249,7 @@ type LatestIssueRun =
|
||||
| "createdAt"
|
||||
> & {
|
||||
resultJson?: unknown;
|
||||
rowVersion?: string;
|
||||
})
|
||||
| null;
|
||||
type SuccessfulLatestIssueRun = NonNullable<LatestIssueRun> & {
|
||||
@@ -977,7 +980,8 @@ export function recoveryService(
|
||||
status: heartbeatRuns.status,
|
||||
error: heartbeatRuns.error,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
contextSnapshot: recoveryRunContextColumn,
|
||||
rowVersion: recoveryRunVersionColumn,
|
||||
livenessState: heartbeatRuns.livenessState,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
startedAt: heartbeatRuns.startedAt,
|
||||
@@ -1007,7 +1011,8 @@ export function recoveryService(
|
||||
status: heartbeatRuns.status,
|
||||
error: heartbeatRuns.error,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
contextSnapshot: recoveryRunContextColumn,
|
||||
rowVersion: recoveryRunVersionColumn,
|
||||
livenessState: heartbeatRuns.livenessState,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
startedAt: heartbeatRuns.startedAt,
|
||||
@@ -1618,7 +1623,10 @@ export function recoveryService(
|
||||
return false;
|
||||
|
||||
const source = await tx
|
||||
.select()
|
||||
.select({
|
||||
...getTableColumns(heartbeatRuns),
|
||||
rowVersion: recoveryRunVersionColumn,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(
|
||||
and(
|
||||
@@ -1634,8 +1642,9 @@ export function recoveryService(
|
||||
source.id !== latestRun.id ||
|
||||
source.agentId !== current.assigneeAgentId ||
|
||||
source.status !== "succeeded" ||
|
||||
JSON.stringify(source.contextSnapshot) !==
|
||||
JSON.stringify(latestRun.contextSnapshot)
|
||||
(latestRun.rowVersion !== undefined
|
||||
? source.rowVersion !== latestRun.rowVersion
|
||||
: JSON.stringify(source.contextSnapshot) !== JSON.stringify(latestRun.contextSnapshot))
|
||||
)
|
||||
return false;
|
||||
const proof = await nativeBlockedUnblockAction(
|
||||
@@ -1903,7 +1912,8 @@ export function recoveryService(
|
||||
status: heartbeatRuns.status,
|
||||
error: heartbeatRuns.error,
|
||||
errorCode: heartbeatRuns.errorCode,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
contextSnapshot: recoveryRunContextColumn,
|
||||
rowVersion: recoveryRunVersionColumn,
|
||||
livenessState: heartbeatRuns.livenessState,
|
||||
resultJson: heartbeatRuns.resultJson,
|
||||
startedAt: heartbeatRuns.startedAt,
|
||||
|
||||
Reference in new issue
Block a user