mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-06 23:20:53 +01:00
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Native tasks save a provider session for later comments and monitor checks. > - The goal projection also records the run that owns the session event sequence. > - A fresh retry uses a new native session, but the projection accepted only the old session. > - The accepted checkpoint then rejected session publication because the goal source still named the old run. > - This pull request lets the current task executor advance that source after a terminal predecessor with no unfinished goal. > - Later turns can then use the fresh retry's saved session. ## Linked Issues or Issue Description Related: #12669, #18343. I searched open and closed PRs for source epochs and native task-session publication. I found no duplicate repair. **What happened?** A successful fresh native retry could retain its predecessor's task-session pointer. A later comment or monitor selected the old provider session. The goal source transition required both runs to have the same native session. The terminal checkpoint then saved a `suppressed_newer` publication result. **Expected behavior** The current task executor can publish its accepted fresh session. A later turn resumes that session. Delayed events cannot restore the old source or revive a cleared goal. **Steps to reproduce** 1. Complete a native turn and retain its task-session row. 2. Admit a fresh session on the same task without deleting the row. 3. Complete that turn, then send an ordinary follow-up. 4. Before this change, the saved pointer can still name the first session. **Paperclip version or commit** Reproduced on source commit `36df8933f0` with a PostgreSQL regression before the production change. **Deployment mode** Source checkout. The affected producer runs for local and remote native sessions. ## What Changed - Permit a different native session to advance the goal source only when its run holds the task execution lock and assignee, and the task is in progress. - Keep the terminal predecessor, native runtime, runner identity, absent or completed goal, source sequence, and conversation reset checks. - Test accepted session publication, a fresh heartbeat followed by an ordinary resumed turn, stale events, current-owner loss, and unfinished or unreadable goals. - Add an opt-in live Codex file-persistence check. Update execution documentation and both recovery maps. ## Verification - Red: the new PostgreSQL publication regression failed before the producer change. The saved session, display ID and goal source still named the predecessor. - Green: 72 distinct tests passed across the final focused runs of `runner-goals.test.ts`, `native-task-session-publication.test.ts`, `native-session-resumption.test.ts`, and `remote-warm-session-recovery.test.ts`. - The assembled heartbeat test admits three turns, opens two provider sessions, resumes the second session once, and stores three results. It uses a deterministic provider backend. - `node scripts/check-maps.mjs` and `git diff --check` passed. - Real local Codex acceptance passed on macOS arm64 with `gpt-6.1-sol` and explicit warm mode. Three heartbeat turns saved three exact file lines. The fresh second turn published its new normalized and provider session IDs. The ordinary third turn retained both IDs. The fixture closed its warm owners and removed its private credential copy and workspace. The production code was at `450e15e`; the final test-only commit selects explicit warm mode and uses the existing owned-home cleanup helper. The production change is identical in the final head. - The earlier per-turn fixture preserved all three file steps and selected the correct normalized session, but provider-thread reuse failed. `run.attach` reported that the Codex provider had pending events; the safe fallback used another provider thread. This remains a separate failure. The first fixture cleanup also failed on immutable instruction assets. The owned-home helper fixed cleanup, and the second per-turn attempt exposed the provider identity assertion directly. - Full current-head CI passes, including typecheck, build, general and serialized tests, Rust/Runner checks, all eight E2E shards, and the `ci / e2e` and `ci / verify` aggregate gates. Greptile is 5/5 on the current commit, with no unresolved review threads. Broad local workspace checks were not duplicated; the focused regressions and actual provider acceptance above were run locally. No merge, deployment, or recovery of an original task is included. ## Risks - Source epoch changes affect saved goal state. The fresh-session path requires current task authority and refuses unfinished or unreadable saved goals. - The change starts no provider work and changes no prior-owner recovery guard. Unknown external outcomes retain their reconciliation hold. - A prior session remains held when it lacks the required authority or state proof. - Per-turn provider-thread reuse remains an open coverage gap described above. This change does not alter that attachment guard. ## Model Used OpenAI Codex, GPT-6 family. The exact model ID and context window are unavailable in this session. The agent used code inspection, tools, and executable tests. ## 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 #` 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>
681 lines
26 KiB
TypeScript
681 lines
26 KiB
TypeScript
import { randomUUID } from "node:crypto";
|
|
import { eq, sql } from "drizzle-orm";
|
|
import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest";
|
|
import {
|
|
agentSessionGoalActions,
|
|
agentTaskSessions,
|
|
agents,
|
|
companies,
|
|
createDb,
|
|
heartbeatRuns,
|
|
issues,
|
|
} from "@paperclipai/db";
|
|
|
|
import {
|
|
getEmbeddedPostgresTestSupport,
|
|
startEmbeddedPostgresTestDatabase,
|
|
} from "../__tests__/helpers/embedded-postgres.js";
|
|
import {
|
|
applyRunnerGoalPrpEvent,
|
|
isRunnerGoalActionCompleted,
|
|
blockRunnerGoalRecovery,
|
|
failRunnerGoalAction,
|
|
RunnerGoalConflictError,
|
|
runnerGoalService,
|
|
} from "./runner-goals.js";
|
|
|
|
import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
|
|
const external = resolveExternalTestDatabaseUrl("PAPERCLIP_RUNNER_GOALS_TEST_DATABASE_URL");
|
|
const embeddedPostgresSupport = external ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
|
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
|
|
|
describeEmbeddedPostgres("runner goal service", () => {
|
|
let db!: ReturnType<typeof createDb>;
|
|
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
|
|
|
beforeAll(async () => {
|
|
if (!external) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-runner-goals-");
|
|
db = createDb(external ?? tempDb!.connectionString);
|
|
}, 20_000);
|
|
|
|
afterEach(async () => {
|
|
await db.delete(agentSessionGoalActions);
|
|
await db.delete(agentTaskSessions);
|
|
await db.delete(issues);
|
|
await db.delete(heartbeatRuns);
|
|
await db.delete(agents);
|
|
await db.delete(companies);
|
|
});
|
|
|
|
afterAll(async () => {
|
|
await db?.$client.end();
|
|
await tempDb?.cleanup();
|
|
});
|
|
|
|
async function seed() {
|
|
const companyId = randomUUID();
|
|
const agentId = randomUUID();
|
|
const issueId = randomUUID();
|
|
await db.insert(companies).values({
|
|
id: companyId,
|
|
name: "Goal Test Company",
|
|
issuePrefix: `G${companyId.replaceAll("-", "").slice(0, 6).toUpperCase()}`,
|
|
requireBoardApprovalForNewAgents: false,
|
|
});
|
|
await db.insert(agents).values({
|
|
id: agentId,
|
|
companyId,
|
|
name: "Codex Goal Agent",
|
|
role: "engineer",
|
|
status: "idle",
|
|
adapterType: "paperclip_runner",
|
|
adapterConfig: {},
|
|
runtimeConfig: {},
|
|
permissions: {},
|
|
});
|
|
await db.insert(issues).values({
|
|
id: issueId,
|
|
companyId,
|
|
identifier: `G-${Math.floor(Math.random() * 1_000_000)}`,
|
|
title: "Pursue a durable goal",
|
|
status: "in_progress",
|
|
assigneeAgentId: agentId,
|
|
});
|
|
return { companyId, agentId, issueId };
|
|
}
|
|
|
|
it.each(["codex_local", "claude_local"])("does not convert %s into another execution path for goals", async (adapterType) => {
|
|
const binding = await seed();
|
|
await db.update(agents).set({ adapterType, adapterConfig: { engine: "acp", mode: "persistent" } });
|
|
const enqueueOfflineControl = vi.fn(async () => {});
|
|
const service = runnerGoalService(db, { enqueueOfflineControl });
|
|
const projection = await service.projection(binding.companyId, binding.issueId);
|
|
expect(projection?.capability).toMatchObject({
|
|
availability: "unsupported",
|
|
reasonCode: "direct_adapter_goal_controller_unavailable",
|
|
actions: [],
|
|
});
|
|
await expect(service.act(binding.companyId, binding.issueId, {
|
|
requestId: randomUUID(), agentId: binding.agentId, expectedRevision: 0,
|
|
action: "create", objective: "Never start an ordinary prompt",
|
|
})).rejects.toThrow();
|
|
expect(enqueueOfflineControl).not.toHaveBeenCalled();
|
|
expect(await db.select().from(agentSessionGoalActions)).toHaveLength(0);
|
|
});
|
|
|
|
it("revisions failed goal starts and ignores repeated failure delivery", async () => {
|
|
const binding = await seed();
|
|
const service = runnerGoalService(db, {
|
|
dispatchLiveControl: () => ({ runId: "run-live", completion: Promise.resolve() }),
|
|
queueLiveCommand: () => null,
|
|
});
|
|
const requestId = randomUUID();
|
|
const accepted = await service.act(binding.companyId, binding.issueId, {
|
|
requestId,
|
|
agentId: binding.agentId,
|
|
expectedRevision: 0,
|
|
action: "create",
|
|
objective: "Observe a failed start",
|
|
});
|
|
const failed = await failRunnerGoalAction(db, {
|
|
...binding, adapterType: "paperclip_runner",
|
|
}, requestId, "provider_start_failed");
|
|
expect(failed?.pendingAction).toBeNull();
|
|
expect(failed?.revision).toBe(accepted.projection.revision + 1);
|
|
expect(await failRunnerGoalAction(db, {
|
|
...binding, adapterType: "paperclip_runner",
|
|
}, requestId, "duplicate_failure")).toBeNull();
|
|
expect((await service.projection(binding.companyId, binding.issueId))?.revision).toBe(failed?.revision);
|
|
});
|
|
|
|
it("enforces revisions, correlates acknowledgements, and fences cleared goals", async () => {
|
|
const binding = await seed();
|
|
const service = runnerGoalService(db, {
|
|
dispatchLiveControl: (_binding, control) => ({
|
|
runId: "run-live",
|
|
completion: Promise.resolve().then(() => undefined),
|
|
}),
|
|
queueLiveCommand: () => null,
|
|
});
|
|
const requestId = randomUUID();
|
|
const accepted = await service.act(binding.companyId, binding.issueId, {
|
|
requestId,
|
|
agentId: binding.agentId,
|
|
expectedRevision: 0,
|
|
action: "create",
|
|
objective: "Finish the goal across turns",
|
|
tokenBudget: 2_000,
|
|
});
|
|
expect(accepted.status).toBe("accepted");
|
|
expect(accepted.projection.pendingAction).toBe("starting");
|
|
expect(await isRunnerGoalActionCompleted(db, binding, requestId)).toBe(false);
|
|
|
|
const repeated = await service.act(binding.companyId, binding.issueId, {
|
|
requestId,
|
|
agentId: binding.agentId,
|
|
expectedRevision: 0,
|
|
action: "create",
|
|
objective: "Finish the goal across turns",
|
|
tokenBudget: 2_000,
|
|
});
|
|
expect(repeated.status).toBe("pending");
|
|
|
|
await expect(service.act(binding.companyId, binding.issueId, {
|
|
requestId,
|
|
agentId: binding.agentId,
|
|
expectedRevision: 0,
|
|
action: "create",
|
|
objective: "Reuse the key for different work",
|
|
tokenBudget: 2_000,
|
|
})).rejects.toMatchObject({
|
|
code: "idempotency_key_conflict",
|
|
projection: { revision: 1 },
|
|
});
|
|
|
|
await expect(service.act(binding.companyId, binding.issueId, {
|
|
requestId: randomUUID(),
|
|
agentId: binding.agentId,
|
|
expectedRevision: 0,
|
|
action: "create",
|
|
objective: "A stale change",
|
|
})).rejects.toBeInstanceOf(RunnerGoalConflictError);
|
|
|
|
await applyRunnerGoalPrpEvent(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
}, {
|
|
eventType: "session.capabilities.updated",
|
|
sourceSeq: 1,
|
|
payload: {
|
|
sessionGoals: {
|
|
availability: "available",
|
|
actions: ["set", "pause", "resume", "clear"],
|
|
autonomousUpdates: true,
|
|
persistentAcrossResume: true,
|
|
maxObjectiveChars: 4_000,
|
|
tokenBudgetControl: true,
|
|
usageReporting: true,
|
|
},
|
|
},
|
|
});
|
|
const updated = await applyRunnerGoalPrpEvent(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
}, {
|
|
eventType: "session.goal.updated",
|
|
sourceSeq: 2,
|
|
payload: {
|
|
requestId,
|
|
workingNow: true,
|
|
goal: {
|
|
objective: "Finish the goal across turns",
|
|
status: "active",
|
|
tokenBudget: 2_000,
|
|
tokensUsed: 25,
|
|
elapsedSeconds: 3,
|
|
iterations: 1,
|
|
createdAt: "2026-08-28T12:00:00.000Z",
|
|
updatedAt: "2026-08-28T12:00:03.000Z",
|
|
completedAt: null,
|
|
},
|
|
},
|
|
});
|
|
expect(updated).toMatchObject({
|
|
goal: { status: "active", workingNow: true },
|
|
pendingAction: null,
|
|
});
|
|
expect(await isRunnerGoalActionCompleted(db, binding, requestId)).toBe(true);
|
|
expect(await isRunnerGoalActionCompleted(db, { ...binding, issueId: randomUUID() }, requestId)).toBe(false);
|
|
expect(await isRunnerGoalActionCompleted(db, { ...binding, agentId: randomUUID() }, requestId)).toBe(false);
|
|
expect(await isRunnerGoalActionCompleted(db, { ...binding, companyId: randomUUID() }, requestId)).toBe(false);
|
|
|
|
const failedClear = await applyRunnerGoalPrpEvent(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
}, {
|
|
eventType: "session.goal.cleared",
|
|
sourceSeq: 3,
|
|
payload: {
|
|
requestId: randomUUID(),
|
|
error: "provider temporarily rejected clear",
|
|
goal: null,
|
|
sessionGoals: {
|
|
availability: "available",
|
|
actions: ["set", "pause", "resume", "clear"],
|
|
autonomousUpdates: true,
|
|
persistentAcrossResume: true,
|
|
maxObjectiveChars: 4_000,
|
|
tokenBudgetControl: true,
|
|
usageReporting: true,
|
|
},
|
|
},
|
|
});
|
|
expect(failedClear).toBeNull();
|
|
const afterFailedClear = await service.projection(
|
|
binding.companyId,
|
|
binding.issueId,
|
|
binding.agentId,
|
|
);
|
|
expect(afterFailedClear).toMatchObject({
|
|
goal: { objective: "Finish the goal across turns", status: "active" },
|
|
});
|
|
|
|
const clearRequestId = randomUUID();
|
|
const clearAccepted = await service.act(binding.companyId, binding.issueId, {
|
|
requestId: clearRequestId,
|
|
agentId: binding.agentId,
|
|
expectedRevision: afterFailedClear!.revision,
|
|
action: "clear",
|
|
});
|
|
expect(clearAccepted.projection.pendingAction).toBe("clearing");
|
|
const cleared = await applyRunnerGoalPrpEvent(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
}, {
|
|
eventType: "session.goal.cleared",
|
|
sourceSeq: 4,
|
|
payload: { requestId: clearRequestId, goal: null, workingNow: false },
|
|
});
|
|
expect(cleared).toMatchObject({ goal: null, pendingAction: null });
|
|
|
|
const stale = await applyRunnerGoalPrpEvent(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
}, {
|
|
eventType: "session.goal.updated",
|
|
sourceSeq: 3,
|
|
payload: {
|
|
goal: { objective: "Stale resurrection", status: "active" },
|
|
},
|
|
});
|
|
expect(stale).toBeNull();
|
|
await expect(service.projection(binding.companyId, binding.issueId, binding.agentId))
|
|
.resolves.toMatchObject({ goal: null });
|
|
});
|
|
|
|
it("accepts a reset source sequence from a successor run of the same durable runner", async () => {
|
|
const binding = await seed();
|
|
const runnerId = randomUUID();
|
|
const nativeSessionId = randomUUID();
|
|
const priorRunId = randomUUID();
|
|
const nextRunId = randomUUID();
|
|
await db.insert(heartbeatRuns).values([
|
|
{ id: priorRunId, companyId: binding.companyId, agentId: binding.agentId, nativeIssueId: binding.issueId, status: "succeeded", invocationSource: "on_demand", runnerInstanceId: runnerId, nativeSessionId },
|
|
{ id: nextRunId, companyId: binding.companyId, agentId: binding.agentId, nativeIssueId: binding.issueId, status: "running", invocationSource: "on_demand", runnerInstanceId: runnerId, nativeSessionId },
|
|
]);
|
|
const first = await applyRunnerGoalPrpEvent(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
}, {
|
|
eventType: "session.goal.updated",
|
|
sourceInstanceId: runnerId,
|
|
sourceRunId: priorRunId,
|
|
sourceSeq: 9,
|
|
payload: {
|
|
goal: {
|
|
objective: "First heartbeat objective",
|
|
status: "complete",
|
|
},
|
|
},
|
|
});
|
|
expect(first).toMatchObject({ goal: { status: "complete" } });
|
|
await expect(applyRunnerGoalPrpEvent(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
}, {
|
|
eventType: "session.goal.snapshot",
|
|
sourceInstanceId: runnerId,
|
|
sourceRunId: priorRunId,
|
|
sourceSeq: 10,
|
|
payload: { goal: null },
|
|
})).resolves.toBeNull();
|
|
await expect(runnerGoalService(db).projection(
|
|
binding.companyId,
|
|
binding.issueId,
|
|
binding.agentId,
|
|
)).resolves.toMatchObject({ revision: 1, goal: { status: "complete" } });
|
|
|
|
const successorEvent = {
|
|
eventType: "session.goal.updated",
|
|
sourceInstanceId: runnerId,
|
|
sourceRunId: nextRunId,
|
|
sourceSeq: 1,
|
|
payload: {
|
|
goal: {
|
|
objective: "Successor heartbeat objective",
|
|
status: "active",
|
|
},
|
|
},
|
|
};
|
|
const eventBinding = { ...binding, adapterType: "paperclip_runner" };
|
|
// Merely changing the source namespace is not proof of succession.
|
|
for (const invalidOwner of [
|
|
{ nativeSessionId: randomUUID() },
|
|
{ nativeIssueId: randomUUID() },
|
|
{ runnerInstanceId: randomUUID() },
|
|
{ status: "succeeded" },
|
|
]) {
|
|
// Corrupt a historical binding to verify the reader's independent guard.
|
|
await db.transaction(async (tx) => {
|
|
await tx.execute(sql`set local session_replication_role = replica`);
|
|
await tx.update(heartbeatRuns).set(invalidOwner).where(eq(heartbeatRuns.id, nextRunId));
|
|
});
|
|
await expect(applyRunnerGoalPrpEvent(db, eventBinding, successorEvent)).resolves.toBeNull();
|
|
await db.transaction(async (tx) => {
|
|
await tx.execute(sql`set local session_replication_role = replica`);
|
|
await tx.update(heartbeatRuns).set({ nativeSessionId, nativeIssueId: binding.issueId, runnerInstanceId: runnerId, status: "running" })
|
|
.where(eq(heartbeatRuns.id, nextRunId));
|
|
});
|
|
}
|
|
await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, priorRunId));
|
|
await expect(applyRunnerGoalPrpEvent(db, eventBinding, successorEvent)).resolves.toBeNull();
|
|
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, priorRunId));
|
|
const successor = await applyRunnerGoalPrpEvent(db, eventBinding, successorEvent);
|
|
expect(successor).toMatchObject({
|
|
goal: { objective: "Successor heartbeat objective", status: "active" },
|
|
});
|
|
|
|
const duplicate = await applyRunnerGoalPrpEvent(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
}, {
|
|
eventType: "session.goal.cleared",
|
|
sourceInstanceId: runnerId,
|
|
sourceRunId: nextRunId,
|
|
sourceSeq: 1,
|
|
payload: { goal: null },
|
|
});
|
|
expect(duplicate).toBeNull();
|
|
const delayed = (eventType: string, sourceSeq: number) => ({
|
|
eventType, sourceInstanceId: runnerId, sourceRunId: priorRunId, sourceSeq,
|
|
payload: { goal: { objective: "Stale predecessor", status: "active" } },
|
|
});
|
|
// The predecessor cannot erase the successor's active goal.
|
|
await expect(applyRunnerGoalPrpEvent(db, eventBinding, delayed("session.goal.cleared", 100))).resolves.toBeNull();
|
|
await expect(runnerGoalService(db).projection(binding.companyId, binding.issueId)).resolves.toMatchObject({
|
|
revision: 2, goal: { objective: "Successor heartbeat objective" },
|
|
});
|
|
await applyRunnerGoalPrpEvent(db, eventBinding, {
|
|
eventType: "session.goal.cleared", sourceInstanceId: runnerId, sourceRunId: nextRunId, sourceSeq: 2, payload: { goal: null },
|
|
});
|
|
// Nor can it resurrect the cleared goal, even once the successor settles.
|
|
await db.update(heartbeatRuns).set({ status: "succeeded" });
|
|
await expect(applyRunnerGoalPrpEvent(db, eventBinding, delayed("session.goal.updated", 101))).resolves.toBeNull();
|
|
await expect(applyRunnerGoalPrpEvent(db, eventBinding, {
|
|
eventType: "session.goal.updated", sourceSeq: 102, payload: delayed("", 0).payload,
|
|
})).resolves.toBeNull();
|
|
await expect(runnerGoalService(db).projection(binding.companyId, binding.issueId)).resolves.toMatchObject({ revision: 3, goal: null });
|
|
const [session] = await db.select().from(agentTaskSessions);
|
|
expect(session).toMatchObject({ goalSourceId: `${runnerId}:${nextRunId}`, goalSourceCursor: 2 });
|
|
});
|
|
|
|
it.each(["current", "current_complete", "not_current", "reassigned", "blocked", "prior_running", "next_finished", "foreign_runner", "missing_session", "unfinished_goal", "unreadable_goal", "conversation_reset"] as const)("fences a fresh goal source by current execution authority (%s)", async kind => {
|
|
const binding = await seed(), priorRunId = randomUUID(), nextRunId = randomUUID();
|
|
const priorRunnerId = randomUUID(), nextRunnerId = randomUUID();
|
|
const priorSessionId = randomUUID(), nextSessionId = randomUUID();
|
|
const eventBinding = { ...binding, adapterType: "paperclip_runner" };
|
|
await db.insert(heartbeatRuns).values([
|
|
{ id: priorRunId, ...binding, nativeIssueId: binding.issueId, runtimeMode: "native", status: "failed",
|
|
runnerInstanceId: priorRunnerId, nativeSessionId: priorSessionId },
|
|
{ id: nextRunId, ...binding, nativeIssueId: binding.issueId, runtimeMode: "native", status: "running",
|
|
runnerInstanceId: nextRunnerId, nativeSessionId: nextSessionId },
|
|
]);
|
|
await db.update(issues).set({ executionRunId: nextRunId }).where(eq(issues.id, binding.issueId));
|
|
const priorEvent = { eventType: "session.goal.updated", sourceInstanceId: priorRunnerId,
|
|
sourceRunId: priorRunId, sourceSeq: 20,
|
|
payload: { goal: { objective: "Retained goal", status: kind === "current_complete" ? "complete" : "active" } } };
|
|
await applyRunnerGoalPrpEvent(db, eventBinding, priorEvent);
|
|
if (kind !== "unfinished_goal" && kind !== "current_complete") await applyRunnerGoalPrpEvent(db, eventBinding, {
|
|
...priorEvent, eventType: "session.goal.cleared", sourceSeq: 21, payload: { goal: null },
|
|
});
|
|
if (kind === "not_current") await db.update(issues).set({ executionRunId: null }).where(eq(issues.id, binding.issueId));
|
|
if (kind === "reassigned") await db.update(issues).set({ assigneeAgentId: null }).where(eq(issues.id, binding.issueId));
|
|
if (kind === "blocked") await db.update(issues).set({ status: "blocked" }).where(eq(issues.id, binding.issueId));
|
|
if (kind === "prior_running") await db.update(heartbeatRuns).set({ status: "running" }).where(eq(heartbeatRuns.id, priorRunId));
|
|
if (kind === "next_finished") await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, nextRunId));
|
|
if (kind === "foreign_runner") await db.update(heartbeatRuns).set({ runnerInstanceId: randomUUID() }).where(eq(heartbeatRuns.id, nextRunId));
|
|
if (kind === "missing_session") await db.update(heartbeatRuns).set({ nativeSessionId: null }).where(eq(heartbeatRuns.id, nextRunId));
|
|
if (kind === "unreadable_goal") await db.update(agentTaskSessions).set({ goalJson: { legacy: "unknown" } }).where(eq(agentTaskSessions.taskKey, binding.issueId));
|
|
if (kind === "conversation_reset") await db.update(issues).set({ conversationAgentId: binding.agentId,
|
|
conversationUserId: "goal-reset-user", conversationState: "active", conversationSessionGeneration: 2,
|
|
}).where(eq(issues.id, binding.issueId));
|
|
const nextEvent = { eventType: "session.goal.updated", sourceInstanceId: nextRunnerId,
|
|
sourceRunId: nextRunId, sourceSeq: 1,
|
|
payload: { goal: { objective: "Current fresh session", status: "active" } } };
|
|
const changed = await applyRunnerGoalPrpEvent(db, eventBinding, nextEvent);
|
|
const [session] = await db.select().from(agentTaskSessions).where(eq(agentTaskSessions.taskKey, binding.issueId));
|
|
if (kind !== "current" && kind !== "current_complete") {
|
|
expect(changed).toBeNull();
|
|
expect(session.goalSourceId).toBe(`${priorRunnerId}:${priorRunId}`);
|
|
expect(session.goalSourceCursor).toBe(kind === "unfinished_goal" ? 20 : 21);
|
|
return;
|
|
}
|
|
expect(changed?.goal).toMatchObject({ objective: "Current fresh session", status: "active" });
|
|
expect(session).toMatchObject({ goalSourceId: `${nextRunnerId}:${nextRunId}`, goalSourceCursor: 1 });
|
|
await applyRunnerGoalPrpEvent(db, eventBinding, { ...nextEvent, eventType: "session.goal.cleared",
|
|
sourceSeq: 2, payload: { goal: null } });
|
|
await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, nextRunId));
|
|
for (const stale of [{ ...priorEvent, sourceSeq: 1_000 }, nextEvent,
|
|
{ eventType: "session.goal.updated", sourceSeq: 2_000, payload: priorEvent.payload }]) {
|
|
expect(await applyRunnerGoalPrpEvent(db, eventBinding, stale)).toBeNull();
|
|
}
|
|
expect(await runnerGoalService(db).projection(binding.companyId, binding.issueId)).toMatchObject({ goal: null });
|
|
expect(await db.select().from(agentSessionGoalActions)).toHaveLength(0);
|
|
});
|
|
|
|
it("keeps a missing resumed goal blocked across restored empty snapshots", async () => {
|
|
const binding = await seed();
|
|
const eventBinding = { ...binding, adapterType: "paperclip_runner" };
|
|
await applyRunnerGoalPrpEvent(db, eventBinding, {
|
|
eventType: "session.goal.updated",
|
|
sourceSeq: 1,
|
|
payload: {
|
|
goal: {
|
|
objective: "Recover the durable provider goal",
|
|
status: "active",
|
|
},
|
|
},
|
|
});
|
|
|
|
const blocked = await applyRunnerGoalPrpEvent(db, eventBinding, {
|
|
eventType: "session.goal.snapshot",
|
|
sourceSeq: 2,
|
|
payload: { goal: null },
|
|
});
|
|
expect(blocked).toMatchObject({
|
|
revision: 2,
|
|
goal: {
|
|
objective: "Recover the durable provider goal",
|
|
status: "blocked",
|
|
lastReason: "provider_session_goal_missing_after_resume",
|
|
},
|
|
});
|
|
|
|
await expect(applyRunnerGoalPrpEvent(db, eventBinding, {
|
|
eventType: "session.goal.snapshot",
|
|
sourceSeq: 3,
|
|
payload: { goal: null },
|
|
})).resolves.toBeNull();
|
|
const [session] = await db.select({
|
|
revision: agentTaskSessions.goalRevision,
|
|
sourceCursor: agentTaskSessions.goalSourceCursor,
|
|
desiredState: agentTaskSessions.goalDesiredState,
|
|
goal: agentTaskSessions.goalJson,
|
|
}).from(agentTaskSessions);
|
|
expect(session).toMatchObject({
|
|
revision: 2,
|
|
sourceCursor: 3,
|
|
desiredState: "paused",
|
|
goal: {
|
|
objective: "Recover the durable provider goal",
|
|
status: "blocked",
|
|
lastReason: "provider_session_goal_missing_after_resume",
|
|
},
|
|
});
|
|
});
|
|
|
|
it("projects committed capability and goal events exactly once across duplicate delivery", async () => {
|
|
const binding = await seed();
|
|
const eventBinding = {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
};
|
|
const source = {
|
|
sourceInstanceId: "native-runner",
|
|
sourceRunId: "goal-heartbeat",
|
|
};
|
|
const capabilityEvent = {
|
|
...source,
|
|
eventType: "session.capabilities.updated",
|
|
sourceSeq: 1,
|
|
payload: {
|
|
sessionGoals: {
|
|
availability: "available",
|
|
actions: ["set", "pause", "resume", "clear"],
|
|
autonomousUpdates: true,
|
|
persistentAcrossResume: true,
|
|
maxObjectiveChars: 4_000,
|
|
tokenBudgetControl: true,
|
|
usageReporting: true,
|
|
},
|
|
},
|
|
};
|
|
const goalEvent = {
|
|
...source,
|
|
eventType: "session.goal.updated",
|
|
sourceSeq: 2,
|
|
payload: {
|
|
workingNow: true,
|
|
goal: {
|
|
objective: "Project the durable provider goal",
|
|
status: "active",
|
|
tokenBudget: 4_000,
|
|
tokensUsed: 50,
|
|
elapsedSeconds: 2,
|
|
iterations: 1,
|
|
},
|
|
},
|
|
};
|
|
|
|
const capability = await applyRunnerGoalPrpEvent(db, eventBinding, capabilityEvent);
|
|
expect(capability).toMatchObject({
|
|
capability: { availability: "available", verified: true },
|
|
revision: 1,
|
|
});
|
|
await expect(
|
|
applyRunnerGoalPrpEvent(db, eventBinding, capabilityEvent),
|
|
).resolves.toBeNull();
|
|
|
|
const goal = await applyRunnerGoalPrpEvent(db, eventBinding, goalEvent);
|
|
expect(goal).toMatchObject({
|
|
goal: {
|
|
objective: "Project the durable provider goal",
|
|
status: "active",
|
|
workingNow: true,
|
|
},
|
|
revision: 2,
|
|
});
|
|
await expect(
|
|
applyRunnerGoalPrpEvent(db, eventBinding, goalEvent),
|
|
).resolves.toBeNull();
|
|
await expect(
|
|
applyRunnerGoalPrpEvent(db, eventBinding, {
|
|
...goalEvent,
|
|
sourceSeq: 3,
|
|
}),
|
|
).resolves.toBeNull();
|
|
|
|
const [session] = await db.select({
|
|
revision: agentTaskSessions.goalRevision,
|
|
sourceCursor: agentTaskSessions.goalSourceCursor,
|
|
capability: agentTaskSessions.goalCapabilityJson,
|
|
goal: agentTaskSessions.goalJson,
|
|
}).from(agentTaskSessions);
|
|
expect(session).toMatchObject({
|
|
revision: 2,
|
|
sourceCursor: 3,
|
|
capability: { availability: "available" },
|
|
goal: {
|
|
objective: "Project the durable provider goal",
|
|
status: "active",
|
|
workingNow: true,
|
|
},
|
|
});
|
|
});
|
|
|
|
it("blocks an unrecoverable active goal with a stable resumable reason", async () => {
|
|
const binding = await seed();
|
|
await applyRunnerGoalPrpEvent(db, { ...binding, adapterType: "paperclip_runner" }, {
|
|
eventType: "session.goal.updated",
|
|
sourceSeq: 1,
|
|
payload: {
|
|
goal: {
|
|
objective: "Recover this goal after restart",
|
|
status: "active",
|
|
tokensUsed: 10,
|
|
elapsedSeconds: 5,
|
|
iterations: 1,
|
|
},
|
|
},
|
|
});
|
|
|
|
const blocked = await blockRunnerGoalRecovery(db, {
|
|
...binding,
|
|
adapterType: "paperclip_runner",
|
|
});
|
|
expect(blocked).toMatchObject({
|
|
goal: {
|
|
status: "blocked",
|
|
workingNow: false,
|
|
lastReason: "provider_session_goal_recovery_failed",
|
|
},
|
|
});
|
|
const [session] = await db.select({
|
|
desired: agentTaskSessions.goalDesiredState,
|
|
}).from(agentTaskSessions);
|
|
expect(session?.desired).toBe("paused");
|
|
});
|
|
|
|
it("routes quiescent goal controls through durable recovery instead of a stale live authority", async () => {
|
|
const binding = await seed();
|
|
let liveDispatches = 0;
|
|
let queuedCommands = 0;
|
|
const offlineControls: string[] = [];
|
|
const service = runnerGoalService(db, {
|
|
dispatchLiveControl: () => {
|
|
liveDispatches += 1;
|
|
return {
|
|
runId: "stale-run",
|
|
completion: Promise.resolve(),
|
|
};
|
|
},
|
|
queueLiveCommand: () => {
|
|
queuedCommands += 1;
|
|
return null;
|
|
},
|
|
enqueueOfflineControl: async ({ requestId }) => {
|
|
offlineControls.push(requestId);
|
|
},
|
|
});
|
|
const requestId = randomUUID();
|
|
|
|
const accepted = await service.act(binding.companyId, binding.issueId, {
|
|
requestId,
|
|
agentId: binding.agentId,
|
|
expectedRevision: 0,
|
|
action: "create",
|
|
objective: "Resume this goal through a durable heartbeat",
|
|
});
|
|
|
|
expect(accepted.status).toBe("accepted");
|
|
expect(liveDispatches).toBe(0);
|
|
expect(queuedCommands).toBe(0);
|
|
expect(offlineControls).toEqual([requestId]);
|
|
await expect(
|
|
db
|
|
.select({ status: agentSessionGoalActions.status })
|
|
.from(agentSessionGoalActions),
|
|
).resolves.toEqual([{ status: "pending" }]);
|
|
});
|
|
});
|