Add guarded task reassignment to the native runner

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-09-19 10:51:36 -05:00
1 parent f589660ec0
commit 00e91d2503
39 files changed
+989 -56

No files matched your search

+19 -1
View File
@@ -3649,6 +3649,7 @@ interface WakeupOptions {
issueStateGuard?: {
statuses: string[];
assigneeAgentId: string;
statusVersion?: number;
};
/** Keep causally distinct external chat continuations out of an existing run. */
allowRunCoalescing?: boolean;
@@ -24072,6 +24073,21 @@ export function heartbeatService(
runnerRemoteProviderPackPath:
runtimeEnv.PAPERCLIP_RUNNER_REMOTE_PROVIDER_PACK_PATH?.trim() ||
null,
stopTaskForReassignment: async (target) => {
await settleLiveRunnerGoalBeforeInterrupt(db, target);
if (!target.runId) return;
const prior = await getRun(target.runId);
if (!prior || prior.companyId !== target.companyId || prior.agentId !== target.agentId) {
throw conflict("Reassignment run binding changed");
}
const stopped = await cancelRunInternal(target.runId, "Cancelled for task reassignment", {
errorCode: "issue_reassigned", suppressImmediateRecovery: true,
resultJson: { reassignmentStopConfirmed: true },
});
if (stopped && ["running", "queued", "scheduled_retry"].includes(stopped.status)) {
throw conflict("The previous run did not stop; reassignment was not applied");
}
},
enqueueWakeup,
onSpawn: async (meta) => {
markDispatchStarted();
@@ -26717,6 +26733,7 @@ export function heartbeatService(
conversationUserId: issues.conversationUserId,
conversationState: issues.conversationState,
status: issues.status,
statusVersion: issues.statusVersion,
projectId: issues.projectId,
projectWorkspaceId: issues.projectWorkspaceId,
executionWorkspaceId: issues.executionWorkspaceId,
@@ -26942,7 +26959,8 @@ export function heartbeatService(
if (
issueStateGuard &&
(!issueStateGuard.statuses.includes(issue.status) ||
issue.assigneeAgentId !== issueStateGuard.assigneeAgentId)
issue.assigneeAgentId !== issueStateGuard.assigneeAgentId ||
(issueStateGuard.statusVersion !== undefined && issue.statusVersion !== issueStateGuard.statusVersion))
) {
await tx.insert(agentWakeupRequests).values({
...durableReceiptFields,
+6
View File
@@ -10819,6 +10819,12 @@ export function issueService(db: Db) {
projectGoalId: nextProjectGoalId,
defaultGoalId: defaultCompanyGoal?.id ?? null,
});
// Ownership changes invalidate observed handoff versions even if status
// stays the same, including an A -> B -> A assignment race.
if ((issueData.assigneeAgentId !== undefined && issueData.assigneeAgentId !== receiptExisting.assigneeAgentId)
|| (issueData.assigneeUserId !== undefined && issueData.assigneeUserId !== receiptExisting.assigneeUserId)) {
patch.statusVersion = sql`${issues.statusVersion} + 1` as unknown as number;
}
// Reasserting Blocked or changing its blockers is a fresh decision even
// when the status string stays the same. Invalidate recovery's prior
// status receipt without treating comment recency as blocking intent.
@@ -6889,6 +6889,7 @@ export async function executePaperclipNativeSession(input: {
runnerRemoteCodexPath?: string | null;
runnerRemoteCodexNpmSpec?: string | null;
runnerRemoteProviderPackPath?: string | null;
stopTaskForReassignment?: (target: { companyId: string; issueId: string; agentId: string; runId: string | null }) => Promise<void>;
enqueueWakeup?: (
agentId: string,
options: {
@@ -9933,6 +9934,7 @@ export async function createRunnerdBackend(input: {
runnerRemoteProviderPackPath?: string | null;
trace?: NativeRunTrace;
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
stopTaskForReassignment?: (target: { companyId: string; issueId: string; agentId: string; runId: string | null }) => Promise<void>;
enqueueWakeup?: (
agentId: string,
options: {
@@ -10048,6 +10050,7 @@ async function createRunnerdBackendWithinSessionClaim(
: undefined,
currentWakeComments: currentWakeComments ?? undefined,
chatAttachmentReadScope: input.chatAttachmentReadScope,
stopTaskForReassignment: input.stopTaskForReassignment,
enqueueWakeup: input.enqueueWakeup,
});
const authorityEpoch = new SessionToolAuthorityEpoch(
@@ -90,7 +90,7 @@ describe("PaperclipRunnerToolAuthority", () => {
issueId,
runId,
});
expect(authority.definitions()).toHaveLength(26);
expect(authority.definitions()).toHaveLength(27);
const questions = authority.definitions().find(tool => tool.name === "request_human_input")!;
expect(questions.description).toContain("ask only the next unanswered question");
expect(questions.description).toContain("Never infer answers");
@@ -1,3 +1,5 @@
import { assertAssignableAgent } from "../agent-assignability.js";
import { authorizationService } from "../authorization.js";
import { handoffPlanContext } from "./handoff-plan-context.js";
import { callCreateSkillTool } from "../skill-tools.js";
import { callProjectTool } from "../project-tools.js";
@@ -23,12 +25,14 @@ import { workspaceFileResourceService } from "../workspace-file-resources.js";
import { badRequest, forbidden } from "../../errors.js";
import { searchRunnerApi } from "./runner-api-catalog.js";
import { executeRunnerApi, validateRunnerApiCall, RUNNER_API_MAX_BYTES, type RunnerApiFile } from "./runner-api-client.js";
import { and, desc, eq, isNull, notInArray } from "drizzle-orm";
import { and, desc, eq, isNull, notInArray, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
activityLog,
agents,
agentWakeupRequests,
chatEndpoints,
chatConversations,
documentRevisions,
heartbeatRuns,
issueApprovals,
@@ -76,7 +80,7 @@ const IMPLEMENTED_OPERATIONS = new Set([
"search_api", "call_api",
"get_task_context", "get_task_history", "search_tasks", "report_progress",
"request_human_input",
"create_skill", "create_task", "set_dependencies", "create_project", "list_project_repositories", "list_projects", "register_deliverable",
"create_skill", "create_task", "reassign_task", "set_dependencies", "create_project", "list_project_repositories", "list_projects", "register_deliverable",
"list_documents", "read_document", "list_document_revisions", "write_document",
"list_agents", "get_agent", "list_approvals", "get_approval", "get_approval_context",
]);
@@ -106,6 +110,7 @@ type Binding = {
readRemoteWorkspaceFile?: RemoteWorkspaceFileReader;
currentWakeComments?: CurrentWakeCommentsBinding;
chatAttachmentReadScope?: NativeChatAttachmentReadScope;
stopTaskForReassignment?: (target: { companyId: string; issueId: string; agentId: string; runId: string | null }) => Promise<void>;
enqueueWakeup?: (agentId: string, options: {
source: "assignment";
triggerDetail: "system";
@@ -115,7 +120,7 @@ type Binding = {
requestedByActorType: "agent";
requestedByActorId: string;
contextSnapshot: Record<string, unknown>;
issueStateGuard?: { statuses: string[]; assigneeAgentId: string };
issueStateGuard?: { statuses: string[]; assigneeAgentId: string; statusVersion?: number };
}) => Promise<unknown>;
};
@@ -467,6 +472,7 @@ export class PaperclipRunnerToolAuthority {
(await captureRunIdentity(this.db, this.binding)).context?.id ?? null);
case "create_task": return this.#createTask(input,
(await captureRunIdentity(this.db, this.binding)).context?.id ?? null);
case "reassign_task": return this.#reassignTask(input);
case "set_dependencies": return this.#setDependencies(input);
case "register_deliverable": return this.#registerDeliverable(input);
default: throw new Error("paperclip_runner_tool_not_bound");
@@ -909,6 +915,131 @@ export class PaperclipRunnerToolAuthority {
return result;
}
async #reassignTask(input: Record<string, unknown>): Promise<unknown> {
const key = requiredString(input.idempotencyKey);
const taskId = requiredUuid(input.taskId);
const assigneeAgentId = requiredUuid(input.assigneeActorId);
const expectedOwner = input.expectedAssigneeActorId === null ? null : requiredUuid(input.expectedAssigneeActorId);
const expectedVersion = input.expectedStatusVersion;
const reason = requiredString(input.reason);
if (!Number.isSafeInteger(expectedVersion) || Number(expectedVersion) < 0 || key.length > 240 || reason.length > 20_000) {
throw new Error("paperclip_runner_tool_input_invalid");
}
if (taskId === this.binding.issueId) {
throw new Error("paperclip_runner_reassignment_current_task: delegate with create_task; this tool cannot interrupt its own caller");
}
const durableKey = `reassign:${this.binding.issueId}:${key}`;
const fingerprint = createHash("sha256").update(canonicalJson(input)).digest("hex");
const priorReceipt = async (tx: Db): Promise<Record<string, unknown> | null> => {
const [prior] = await tx.select({ details: activityLog.details }).from(activityLog).where(and(
eq(activityLog.companyId, this.binding.companyId), eq(activityLog.action, "issue.reassigned"),
sql`${activityLog.details}->>'reassignmentKey' = ${durableKey}`,
)).limit(1);
if (!prior) return null;
if (prior.details?.inputFingerprint !== fingerprint) throw new Error("paperclip_runner_tool_idempotency_conflict");
return record(prior.details?.receipt);
};
const authorize = async (tx: Db, run: typeof heartbeatRuns.$inferSelect) => {
const [target] = await tx.select().from(issues).where(and(eq(issues.id, taskId), eq(issues.companyId, this.binding.companyId))).for("update");
if (!target) throw new Error("paperclip_runner_task_not_found");
const actor = { type: "agent" as const, source: "agent_jwt" as const, companyId: this.binding.companyId,
agentId: this.binding.agentId, runId: this.binding.runId, onBehalfOfUserId: run.responsibleUserId };
for (const action of ["issue:read", "tasks:assign"] as const) {
const decision = await authorizationService(tx).decide({ actor, action, resource: {
type: "issue", companyId: this.binding.companyId, issueId: taskId, projectId: target.projectId,
parentIssueId: target.parentId, assigneeAgentId: action === "tasks:assign" ? assigneeAgentId : target.assigneeAgentId,
assigneeUserId: action === "tasks:assign" ? null : target.assigneeUserId, status: target.status,
} });
if (!decision.allowed) throw forbidden(decision.explanation);
}
await assertAssignableAgent(tx, this.binding.companyId, assigneeAgentId, { kind: "work" });
const [externalChat] = await tx.select({ id: chatConversations.id }).from(chatConversations).where(and(
eq(chatConversations.companyId, this.binding.companyId), eq(chatConversations.issueId, taskId),
)).limit(1);
if (externalChat) throw new Error("paperclip_runner_reassignment_conversation_locked");
return target;
};
const validate = (target: typeof issues.$inferSelect) => {
if (target.conversationAgentId) throw new Error("paperclip_runner_reassignment_conversation_locked");
if (["done", "cancelled", "in_review"].includes(target.status) || record(target.executionState).status === "pending") {
throw new Error("paperclip_runner_reassignment_state_denied");
}
if (target.assigneeUserId || target.assigneeAgentId !== expectedOwner || target.statusVersion !== expectedVersion) {
throw new Error("paperclip_runner_reassignment_conflict: refresh search_tasks before deciding again");
}
};
// Stop outside row locks: the runtime needs those locks to acknowledge cancellation.
// A second locked check below prevents any intervening ownership/version change.
const prepared = await this.db.transaction(async (tx) => {
const context = await this.#lockAuthorizedMutationContext(tx as unknown as Db);
if (context.issue.workMode !== "standard") throw new Error("paperclip_runner_tool_mode_denied");
const target = await authorize(tx as unknown as Db, context.run);
if (await priorReceipt(tx as unknown as Db)) return null;
validate(target);
return target;
});
if (prepared && prepared.assigneeAgentId && prepared.assigneeAgentId !== assigneeAgentId) {
if (!this.binding.stopTaskForReassignment && prepared.executionRunId) {
throw new Error("paperclip_runner_reassignment_stop_unavailable");
}
await this.binding.stopTaskForReassignment?.({ companyId: this.binding.companyId, issueId: taskId,
agentId: prepared.assigneeAgentId, runId: prepared.executionRunId });
}
let publication: Awaited<ReturnType<typeof persistActivity>>["publication"] | null = null;
const result = await this.#withMutationReceipt("reassign_task", key, input, async (tx, context) => {
if (context.issue.workMode !== "standard") throw new Error("paperclip_runner_tool_mode_denied");
const target = await authorize(tx, context.run);
const prior = await priorReceipt(tx);
if (prior) return prior;
validate(target);
if (target.executionRunId && target.executionRunId !== prepared?.executionRunId) {
throw new Error("paperclip_runner_reassignment_conflict: a new execution started; refresh before retrying");
}
if (target.executionRunId && target.assigneeAgentId !== assigneeAgentId) {
const [run] = await tx.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, target.executionRunId));
if (run && ["running", "queued", "scheduled_retry"].includes(run.status)) throw new Error("paperclip_runner_reassignment_stop_unconfirmed");
}
const changed = target.assigneeAgentId !== assigneeAgentId;
const updated = changed ? await issueService(tx).update(taskId, {
companyGuard: this.binding.companyId, assigneeAgentId, assigneeUserId: null,
status: target.status === "in_progress" ? "todo" : target.status,
statusVersion: target.statusVersion + 1, actorAgentId: this.binding.agentId,
}, tx) : target;
if (!updated) throw new Error("paperclip_runner_task_not_found");
const receipt = {
commandId: `reassign-task:${taskId}:${updated.statusVersion}`, disposition: changed ? "applied" : "duplicate",
stateRevision: updated.statusVersion, entityRefs: [taskId],
scheduledWakeIds: changed && updated.status === "todo" ? [`${durableKey}:${updated.statusVersion}`] : [],
};
if (changed) await issueService(tx).addComment(taskId, reason, { agentId: this.binding.agentId, runId: this.binding.runId });
const activity = await persistActivity(tx, {
companyId: this.binding.companyId, actorType: "agent", actorId: this.binding.agentId,
agentId: this.binding.agentId, runId: this.binding.runId, issueId: taskId,
action: "issue.reassigned", entityType: "issue", entityId: taskId,
details: { source: "paperclip_runner_protocol", reassignmentKey: durableKey, inputFingerprint: fingerprint,
previousAssigneeAgentId: target.assigneeAgentId, assigneeAgentId, reason, changed, receipt },
});
publication = activity.publication;
return receipt;
}, { beforeReceiptReplay: async (tx, context) => { await authorize(tx, context.run); } }) as {
stateRevision: number; scheduledWakeIds: string[];
};
if (publication) publishActivity(publication);
// Replay repairs a failed dispatch. The durable wake key and state/version guard
// prevent duplicate successors and prevent waking a later, unrelated assignment.
if (result.scheduledWakeIds.length && this.binding.enqueueWakeup) {
await this.binding.enqueueWakeup(assigneeAgentId, {
source: "assignment", triggerDetail: "system", reason: "issue_assigned",
payload: { issueId: taskId, mutation: "reassign_task", reason },
idempotencyKey: result.scheduledWakeIds[0]!, requestedByActorType: "agent", requestedByActorId: this.binding.agentId,
contextSnapshot: { issueId: taskId, source: "paperclip_runner.reassign_task", reassignmentReason: reason },
issueStateGuard: { statuses: ["todo"], assigneeAgentId, statusVersion: result.stateRevision },
});
}
return result;
}
async #setDependencies(input: Record<string, unknown>): Promise<unknown> {
const idempotencyKey = requiredString(input.idempotencyKey);
if (!Array.isArray(input.blockedByTaskIds)) {
@@ -48,7 +48,7 @@ function dedicatedTools(method: string, path: string): string[] {
if (/\/issues\/\{[^}]+\}\/comments$/.test(path)) return method === "GET" ? ["get_task_history"] : ["report_progress"];
if (/\/issues\/\{[^}]+\}\/documents/.test(path)) return method === "DELETE" ? [] : method === "GET" ? ["list_documents", "read_document", "list_document_revisions"] : ["write_document"];
if (/\/issues$/.test(path)) return method === "GET" ? ["search_tasks"] : ["create_task"];
if (/\/issues\/\{[^}]+\}$/.test(path)) return method === "GET" ? ["get_task_context"] : ["set_dependencies", "finish_task", "block_task", "request_review"];
if (/\/issues\/\{[^}]+\}$/.test(path)) return method === "GET" ? ["get_task_context"] : ["reassign_task", "set_dependencies", "finish_task", "block_task", "request_review"];
if (/\/agents$/.test(path) && method === "GET") return ["list_agents"];
if (/\/agents\/(me|\{[^}]+\})$/.test(path) && method === "GET") return ["get_agent"];
if (/\/approvals$/.test(path) && method === "GET") return ["list_approvals"];
@@ -0,0 +1,165 @@
import { randomUUID } from "node:crypto";
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
import { eq, and } from "drizzle-orm";
import { activityLog, agents, companies, createDb, heartbeatRuns, issues, issueComments } from "@paperclipai/db";
import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
import { issueService } from "../issues.js";
import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js";
describe("runner task reassignment", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: ReturnType<typeof createDb>;
beforeAll(async () => {
temporary = await startEmbeddedPostgresTestDatabase("runner-reassign-");
db = createDb(temporary.connectionString);
});
afterAll(async () => { await temporary?.cleanup(); });
async function fixture(status = "todo") {
const companyId = randomUUID(), agentId = randomUUID(), nextId = randomUUID(), issueId = randomUUID(), targetId = randomUUID(), runId = randomUUID();
await db.insert(companies).values({ id: companyId, name: "Reassignment", issuePrefix: `R${companyId.slice(0, 6)}` });
await db.insert(agents).values([
{ id: agentId, companyId, name: "Manager", role: "ceo", status: "active", adapterType: "paperclip_runner" },
{ id: nextId, companyId, name: "Engineer", role: "engineer", status: "active", adapterType: "paperclip_runner" },
]);
await db.insert(issues).values([
{ id: issueId, companyId, title: "Coordinate", status: "in_progress", assigneeAgentId: agentId },
{ id: targetId, companyId, title: "Existing work", status, assigneeAgentId: agentId, description: "Keep the existing plan and context" },
]);
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: issueId, invocationSource: "assignment", triggerDetail: "system", contextSnapshot: { issueId } });
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId));
const enqueueWakeup = vi.fn().mockResolvedValue({ id: randomUUID() });
const stopTaskForReassignment = vi.fn().mockResolvedValue(undefined);
const binding = { companyId, agentId, issueId, runId, enqueueWakeup, stopTaskForReassignment };
const authority = new PaperclipRunnerToolAuthority(db, binding);
const call = { tool: "reassign_task", callId: "handoff", arguments: { taskId: targetId, assigneeActorId: nextId, expectedAssigneeActorId: agentId, expectedStatusVersion: 0, reason: "Engineer owns implementation; retain the existing plan.", idempotencyKey: "handoff" } };
const target = async () => (await db.select().from(issues).where(eq(issues.id, targetId)))[0]!;
return { companyId, agentId, nextId, issueId, targetId, runId, authority, binding, call, target, enqueueWakeup, stopTaskForReassignment };
}
it("moves existing work once, persists context and audit, and dispatches a guarded wake", async () => {
const f = await fixture();
const receipt = await f.authority.execute(f.call);
expect(await f.target()).toMatchObject({ assigneeAgentId: f.nextId, status: "todo", statusVersion: 1, description: "Keep the existing plan and context" });
await expect(f.authority.execute({ ...f.call, callId: "retry" })).resolves.toEqual(receipt);
expect(f.stopTaskForReassignment).toHaveBeenCalledTimes(1);
expect(f.enqueueWakeup).toHaveBeenCalledWith(f.nextId, expect.objectContaining({ issueStateGuard: { statuses: ["todo"], assigneeAgentId: f.nextId, statusVersion: 1 } }));
expect(f.enqueueWakeup.mock.calls[0]![1].idempotencyKey).toEqual(f.enqueueWakeup.mock.calls[1]![1].idempotencyKey);
expect(await db.select().from(activityLog).where(and(eq(activityLog.companyId, f.companyId), eq(activityLog.action, "issue.reassigned")))).toHaveLength(1);
expect(await db.select().from(issueComments).where(eq(issueComments.issueId, f.targetId))).toHaveLength(1);
});
it.each(["blocked", "backlog"])("preserves %s and does not wake work prematurely", async status => {
const f = await fixture(status);
await f.authority.execute(f.call);
expect(await f.target()).toMatchObject({ assigneeAgentId: f.nextId, status });
expect(f.enqueueWakeup).not.toHaveBeenCalled();
});
it("replays across a replacement caller run without a second mutation", async () => {
const f = await fixture();
const receipt = await f.authority.execute(f.call);
const replacement = randomUUID();
await db.insert(heartbeatRuns).values({ id: replacement, companyId: f.companyId, agentId: f.agentId, nativeIssueId: f.issueId, runtimeMode: "native", status: "running", invocationSource: "assignment", triggerDetail: "system" });
await db.update(issues).set({ executionRunId: replacement }).where(eq(issues.id, f.issueId));
await expect(new PaperclipRunnerToolAuthority(db, { ...f.binding, runId: replacement }).execute(f.call)).resolves.toEqual(receipt);
expect((await f.target()).statusVersion).toBe(1);
});
it("repairs a dispatch failure by replaying the committed receipt", async () => {
const f = await fixture();
f.enqueueWakeup.mockRejectedValueOnce(new Error("queue unavailable"));
await expect(f.authority.execute(f.call)).rejects.toThrow("queue unavailable");
expect((await f.target()).assigneeAgentId).toBe(f.nextId);
await f.authority.execute(f.call);
expect(f.stopTaskForReassignment).toHaveBeenCalledTimes(1);
expect(f.enqueueWakeup).toHaveBeenCalledTimes(2);
});
it("rejects key reuse with different input and stale ownership/version", async () => {
const f = await fixture();
await f.authority.execute(f.call);
await expect(f.authority.execute({ ...f.call, arguments: { ...f.call.arguments, reason: "Changed intent" } })).rejects.toThrow("idempotency_conflict");
await expect(f.authority.execute({ ...f.call, arguments: { ...f.call.arguments, idempotencyKey: "stale" } })).rejects.toThrow("reassignment_conflict");
});
it("stops active execution before transferring ownership", async () => {
const f = await fixture("in_progress");
const oldRun = randomUUID();
await db.insert(heartbeatRuns).values({ id: oldRun, companyId: f.companyId, agentId: f.agentId, nativeIssueId: f.targetId, runtimeMode: "native", status: "running", invocationSource: "assignment", triggerDetail: "system" });
await db.update(issues).set({ executionRunId: oldRun, checkoutRunId: oldRun }).where(eq(issues.id, f.targetId));
f.stopTaskForReassignment.mockImplementationOnce(async () => {
expect((await f.target()).assigneeAgentId).toBe(f.agentId);
await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, oldRun));
});
await f.authority.execute(f.call);
expect(await f.target()).toMatchObject({ assigneeAgentId: f.nextId, status: "todo", executionRunId: null, checkoutRunId: null });
});
it("does not transfer ownership if stop fails or a new run races the stop", async () => {
const f = await fixture();
f.stopTaskForReassignment.mockRejectedValueOnce(new Error("stop failed"));
await expect(f.authority.execute(f.call)).rejects.toThrow("stop failed");
expect((await f.target()).assigneeAgentId).toBe(f.agentId);
f.stopTaskForReassignment.mockImplementationOnce(async () => {
const runId = randomUUID();
await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: f.agentId, status: "running", invocationSource: "assignment", triggerDetail: "system" });
await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, f.targetId));
});
await expect(f.authority.execute(f.call)).rejects.toThrow("reassignment_conflict");
expect(f.enqueueWakeup).not.toHaveBeenCalled();
});
it.each(["done", "cancelled", "in_review"])("rejects %s before interrupting work", async status => {
const f = await fixture(status);
await expect(f.authority.execute(f.call)).rejects.toThrow("state_denied");
expect(f.stopTaskForReassignment).not.toHaveBeenCalled();
});
it("rejects conversations and the caller’s own task", async () => {
const f = await fixture();
await db.update(issues).set({ conversationAgentId: f.agentId, conversationUserId: "test-user", conversationState: "active", status: "in_progress" }).where(eq(issues.id, f.targetId));
await expect(f.authority.execute(f.call)).rejects.toThrow("conversation_locked");
await expect(f.authority.execute({ ...f.call, arguments: { ...f.call.arguments, taskId: f.issueId } })).rejects.toThrow("current_task");
expect(f.stopTaskForReassignment).not.toHaveBeenCalled();
});
it.each(["planning", "ask"])("denies reassignment from %s mode", async workMode => {
const f = await fixture();
await db.update(issues).set({ workMode }).where(eq(issues.id, f.issueId));
await expect(f.authority.execute(f.call)).rejects.toThrow("mode_denied");
});
it("denies foreign targets and unavailable assignees", async () => {
const f = await fixture(), foreign = await fixture();
await expect(f.authority.execute({ ...f.call, arguments: { ...f.call.arguments, taskId: foreign.targetId } })).rejects.toThrow("task_not_found");
await expect(f.authority.execute({ ...f.call, arguments: { ...f.call.arguments, assigneeActorId: foreign.agentId } })).rejects.toThrow();
await db.update(agents).set({ status: "pending_approval" }).where(eq(agents.id, f.nextId));
await expect(f.authority.execute(f.call)).rejects.toThrow();
expect(f.stopTaskForReassignment).not.toHaveBeenCalled();
});
it("rejects an ownership ABA race through the existing issue service", async () => {
const f = await fixture();
await issueService(db).update(f.targetId, { assigneeAgentId: f.nextId });
await issueService(db).update(f.targetId, { assigneeAgentId: f.agentId });
expect((await f.target()).statusVersion).toBe(2);
await expect(f.authority.execute(f.call)).rejects.toThrow("reassignment_conflict");
expect(f.stopTaskForReassignment).not.toHaveBeenCalled();
});
it("respects protected assignment policies instead of inheriting board authority", async () => {
const f = await fixture();
await db.update(agents).set({ permissions: { authorizationPolicy: { protectedAgent: { blockAssignment: true } } } }).where(eq(agents.id, f.nextId));
await expect(f.authority.execute(f.call)).rejects.toThrow();
expect((await f.target()).assigneeAgentId).toBe(f.agentId);
expect(f.stopTaskForReassignment).not.toHaveBeenCalled();
});
it("rejects writes after the caller loses its run binding", async () => {
const f = await fixture();
await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, f.runId));
await expect(f.authority.execute(f.call)).rejects.toThrow("binding_not_authorized");
expect((await f.target()).assigneeAgentId).toBe(f.agentId);
});
});