fix(runner): restore task runtime parity (#12685)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - The task view shows a running agent and lets an operator guide that
agent.
> - The merged runner stack lost parts of the accepted task experience.
> - Native event errors could hide current reasoning from the operator.
> - Queued message steering had no server route on `master`.
> - This pull request restores the task-runtime behavior and keeps the
runner experimental gate.
> - The benefit is a visible and steerable native run with durable
fallback behavior.

## Linked Issues or Issue Description

**What happened?**

The task view could stop showing current runner reasoning. The steering
action also failed because the server route was absent. Runner
instruction files were not declared as supported.

**Expected behavior**

The task view must show current provider activity. It must use the live
log when durable native events are empty or unavailable. The operator
must be able to steer a queued message into the active native turn.

**Steps to reproduce**

1. Enable the Paperclip Runner experimental setting.
2. Start a native runner task.
3. Open the task view while the run emits reasoning.
4. Queue a message and select the steering action.

**Paperclip version or commit**

The regression reproduces on `24a674f8858060e77ea1beb50689d26473e91431`.

**Additional context**

Related closed work: Refs #12592.

## What Changed

- Restored the queued-comment steering route for active native sessions.
- Added durable and queue-bound steering acknowledgements for safe
retries.
- Restored runner instruction bundle support.
- Added live-log fallback when native events are empty or unavailable.
- Restored the compact live reasoning ticker in the task view.
- Added a visible temporary-unavailable state when both activity sources
fail.
- Kept the unified Paperclip Runner experimental gate unchanged.

## Verification

- `pnpm --filter @paperclipai/ui typecheck`
- `pnpm --filter @paperclipai/server typecheck`
- `pnpm -r typecheck`
- `pnpm check:token-gates`
- `pnpm build`
- Seven focused test files passed with 140 tests.
- The final steering regression file passed with 12 tests.
- The broad local test run reached unrelated workspace, port, and shared
database failures. The changed-area tests remained green.

## Risks

- The steering route changes queue and run records in one transaction.
Tests cover stale targets, unavailable sessions, lost responses, and
wrong-queue acknowledgements.
- Native events remain the primary transcript source. The live log is
used only when event data is absent or its poll fails.
- The experimental gate still hides and rejects the runner when the
setting is off.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

OpenAI Codex, GPT-5, with tool use, code execution, and subagent review.

## 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
- [ ] I have updated relevant documentation to reflect my changes — no
documentation change is required for this regression repair
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
This commit is contained in:
Dotta authored and GitHub committed 2026-09-01 21:17:54 -05:00
1 parent b4f302d040
commit 72b9f92d76
14 files changed
+884 -34

No files matched your search

@@ -233,6 +233,8 @@ describe("server adapter registry", () => {
it("rejects an unsupported persisted runner provider before probing Codex", async () => {
const adapter = requireServerAdapter("paperclip_runner");
expect(adapter.supportsInstructionsBundle).toBe(true);
expect(adapter.instructionsPathKey).toBe("instructionsFilePath");
const result = await adapter.testEnvironment({
companyId: "company-1",
adapterType: "paperclip_runner",
+1 -1
View File
@@ -163,7 +163,7 @@ describe("adapter routes", () => {
.toMatchObject({
disabled: false,
capabilities: {
supportsInstructionsBundle: false,
supportsInstructionsBundle: true,
},
});
});
@@ -596,6 +596,11 @@ describe("agent routes adapter validation", () => {
expect(res.status, JSON.stringify(res.body)).toBe(201);
expect(mockAgentService.create).toHaveBeenCalledOnce();
expect(mockAgentInstructionsService.materializeManagedBundle).toHaveBeenCalledWith(
expect.objectContaining({ adapterType: "paperclip_runner" }),
expect.any(Object),
expect.objectContaining({ entryFile: "AGENTS.md", replaceExisting: false }),
);
});
it("rejects non-Codex providers on fresh paperclip_runner agents and hires", async () => {
@@ -412,6 +412,170 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
expect(discard.status).toBe(403);
});
it("leaves the selected row queued when no native steering session is attached", async () => {
const seeded = await seedQueue();
const initial = await request(app(seeded.companyId))
.get(`/api/issues/${seeded.issueId}/queued-comments`);
expect(initial.body.steeringDisposition).toBe("temporarily_unavailable");
const steered = await request(app(seeded.companyId))
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
.send({ queueId: seeded.wakeId, targetRunId: seeded.runId, revision: initial.body.revision });
expect(steered.status).toBe(409);
expect(steered.body.details).toMatchObject({
code: "steering_temporarily_unavailable",
retryable: true,
});
const queueAfterFailure = await request(app(seeded.companyId))
.get(`/api/issues/${seeded.issueId}/queued-comments`);
expect(queueAfterFailure.body.entries.map((entry: any) => entry.comment.id)).toEqual(seeded.commentIds);
});
it("returns the persisted acknowledgement when the final steering response is retried", async () => {
const seeded = await seedQueue();
await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1]));
await db
.update(agentWakeupRequests)
.set({
status: "cancelled",
finishedAt: new Date("2026-08-22T15:04:00.000Z"),
payload: {
issueId: seeded.issueId,
commentId: seeded.commentIds[0],
_paperclipWakeContext: {
commentId: seeded.commentIds[0],
wakeCommentId: seeded.commentIds[0],
wakeCommentIds: [seeded.commentIds[0]],
},
},
})
.where(eq(agentWakeupRequests.id, seeded.wakeId));
await db
.update(heartbeatRuns)
.set({
resultJson: {
queuedSteeringAcknowledgements: {
[seeded.commentIds[0]]: {
status: "acknowledged",
queueId: seeded.wakeId,
turnId: "turn-acknowledged",
acknowledgedAt: "2026-08-22T15:04:00.000Z",
},
},
},
})
.where(eq(heartbeatRuns.id, seeded.runId));
const retried = await request(app(seeded.companyId))
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
.send({
queueId: seeded.wakeId,
targetRunId: seeded.runId,
revision: "response-was-lost-before-the-client-stored-the-revision",
});
expect(retried.status, JSON.stringify(retried.body)).toBe(200);
expect(retried.body).toMatchObject({
issueId: seeded.issueId,
queueId: null,
state: null,
entries: [],
});
const activity = await db
.select({ details: activityLog.details })
.from(activityLog)
.where(eq(activityLog.action, "issue.queued_comment_steered"))
.then((rows) => rows[0]);
expect(activity?.details).toMatchObject({
commentId: seeded.commentIds[0],
targetRunId: seeded.runId,
turnId: "turn-acknowledged",
duplicate: true,
});
});
it("does not reuse an acknowledgement from a different queue", async () => {
const seeded = await seedQueue();
const initial = await request(app(seeded.companyId))
.get(`/api/issues/${seeded.issueId}/queued-comments`);
await db
.update(heartbeatRuns)
.set({
resultJson: {
queuedSteeringAcknowledgements: {
[seeded.commentIds[0]]: {
status: "acknowledged",
queueId: randomUUID(),
turnId: "turn-from-another-queue",
acknowledgedAt: "2026-08-22T15:04:00.000Z",
},
},
},
})
.where(eq(heartbeatRuns.id, seeded.runId));
const steered = await request(app(seeded.companyId))
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
.send({
queueId: seeded.wakeId,
targetRunId: seeded.runId,
revision: initial.body.revision,
});
expect(steered.status).toBe(409);
expect(steered.body.details).toMatchObject({
code: "steering_temporarily_unavailable",
retryable: true,
});
const queueAfterFailure = await request(app(seeded.companyId))
.get(`/api/issues/${seeded.issueId}/queued-comments`);
expect(queueAfterFailure.body.entries.map((entry: any) => entry.comment.id))
.toEqual(seeded.commentIds);
});
it("keeps queue edits available during handoff but rejects stale same-turn steering", async () => {
const seeded = await seedQueue();
const initial = await request(app(seeded.companyId))
.get(`/api/issues/${seeded.issueId}/queued-comments`);
await db
.update(heartbeatRuns)
.set({ status: "succeeded", finishedAt: new Date("2026-08-22T15:05:00.000Z") })
.where(eq(heartbeatRuns.id, seeded.runId));
const edit = await request(app(seeded.companyId))
.patch(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}`)
.send({
queueId: seeded.wakeId,
revision: initial.body.revision,
body: "edited during handoff",
});
expect(edit.status, JSON.stringify(edit.body)).toBe(200);
const stored = await db
.select({ body: issueComments.body })
.from(issueComments)
.where(eq(issueComments.id, seeded.commentIds[0]))
.then((rows) => rows[0]);
expect(stored?.body).toBe("edited during handoff");
const steer = await request(app(seeded.companyId))
.post(`/api/issues/${seeded.issueId}/queued-comments/${seeded.commentIds[0]}/steer`)
.send({
queueId: seeded.wakeId,
targetRunId: seeded.runId,
revision: edit.body.revision,
});
expect(steer.status).toBe(409);
expect(steer.body.details?.code).toBe("queued_comment_stale_target");
const wake = await db
.select({ status: agentWakeupRequests.status, payload: agentWakeupRequests.payload })
.from(agentWakeupRequests)
.where(eq(agentWakeupRequests.id, seeded.wakeId))
.then((rows) => rows[0]);
expect(wake?.status).toBe("deferred_issue_execution");
expect((wake?.payload as any)?._paperclipWakeContext?.wakeCommentIds).toEqual(seeded.commentIds);
});
it("cancels a queued continuation whose comments disappeared before claim", async () => {
const seeded = await seedQueue();
const queueRunId = await promoteQueue(seeded);
@@ -437,6 +601,27 @@ describeEmbeddedPostgres("issue queued-comment routes", () => {
expect(wake?.status).toBe("cancelled");
});
it("keeps a persisted legacy queue on the legacy protocol after the agent changes adapters", async () => {
const seeded = await seedQueue();
const queueRunId = await promoteQueue(seeded);
await db
.update(heartbeatRuns)
.set({ runtimeMode: "legacy" })
.where(eq(heartbeatRuns.id, queueRunId));
const queued = await request(app(seeded.companyId))
.get(`/api/issues/${seeded.issueId}/queued-comments`);
expect(queued.status, JSON.stringify(queued.body)).toBe(200);
expect(queued.body).toMatchObject({
queueId: seeded.wakeId,
state: "queued",
targetRunId: null,
protocol: "legacy",
steeringDisposition: "unsupported",
});
});
it("serializes discard against queued-run claim", async () => {
const seeded = await seedQueue();
await db.delete(issueComments).where(eq(issueComments.id, seeded.commentIds[1]));
+2 -1
View File
@@ -419,7 +419,8 @@ const paperclipRunnerAdapter: ServerAdapterModule = {
listModels: listCodexModels,
refreshModels: refreshCodexModels,
supportsLocalAgentJwt: false,
supportsInstructionsBundle: false,
supportsInstructionsBundle: true,
instructionsPathKey: "instructionsFilePath",
requiresMaterializedRuntimeSkills: false,
getRuntimeCommandSpec: (config) => buildNpmRuntimeCommandSpec(config, "codex", "@openai/codex"),
agentConfigurationDoc:
+236 -3
View File
@@ -263,6 +263,11 @@ import {
observeCrossIssueInfluence,
type CrossIssueInfluenceKind,
} from "../services/cross-issue-influence-limit.js";
import {
getNativeSessionSteeringState,
NativeSessionSteeringError,
steerNativeSession,
} from "../services/native-runtime/native-session-executor.js";
import {
queuedCommentIdsFromWakePayload,
withQueuedCommentIdsInRunContext,
@@ -277,6 +282,9 @@ const queuedCommentMutationTargetSchema = z.object({
queueId: z.string().min(1),
revision: z.string().min(1),
});
const queuedCommentSteeringTargetSchema = queuedCommentMutationTargetSchema.extend({
targetRunId: z.string().min(1),
});
const editQueuedCommentSchema = queuedCommentMutationTargetSchema.extend({
body: z
.string()
@@ -5511,12 +5519,33 @@ export function issueRoutes(
: input.queueState;
const wake = queueState?.wake ?? null;
const comments = await queueCommentsForWake(input.executor, input.issue.id, wake);
const protocol = input.activeRun?.runtimeMode === "native"
|| queueState?.queueRun?.runtimeMode === "native"
const assignedAgent = input.issue.assigneeAgentId
? await input.executor
.select({ adapterType: agents.adapterType })
.from(agents)
.where(and(
eq(agents.id, input.issue.assigneeAgentId),
eq(agents.companyId, input.issue.companyId),
))
.limit(1)
.then((rows) => rows[0] ?? null)
: null;
const persistedRuntimeMode = queueState?.state === "queued" && queueState.queueRun
? queueState.queueRun.runtimeMode
: queueState?.state === "deferred" && input.activeRun
? input.activeRun.runtimeMode
: null;
const protocol = persistedRuntimeMode === "native"
|| (persistedRuntimeMode === null && assignedAgent?.adapterType === "paperclip_runner")
? "paperclip_runner_v1" as const
: "legacy" as const;
const steeringRun = queueState?.state === "deferred" ? input.activeRun : null;
let steeringDisposition = input.steeringDisposition ?? "unsupported" as const;
let steeringDisposition = input.steeringDisposition
?? (protocol === "paperclip_runner_v1" && steeringRun
? await getNativeSessionSteeringState(steeringRun.id)
.then((state) => state.disposition)
.catch(() => "temporarily_unavailable" as const)
: "unsupported" as const);
if (protocol === "paperclip_runner_v1" && (!steeringRun || comments.length === 0)) {
steeringDisposition = "temporarily_unavailable";
}
@@ -11851,6 +11880,210 @@ export function issueRoutes(
},
);
router.post(
"/issues/:id/queued-comments/:commentId/steer",
validate(queuedCommentSteeringTargetSchema),
async (req, res) => {
assertBoard(req);
if (!req.actor.userId) throw forbidden("Board user context required");
const id = req.params.id as string;
const commentId = req.params.commentId as string;
const issue = await getAccessibleResource(req, res, svc.getById(id), "Issue not found");
if (!issue) return;
const actor = getActorInfo(req);
let acknowledgedTurnId: string | null = null;
let duplicate = false;
let queue: IssueQueuedCommentQueue;
try {
queue = await db.transaction(async (tx) => {
// A client can lose the successful response after the final queued
// message cancels its wake. Lock the original queue and target run
// first so that the persisted acknowledgement remains a durable
// idempotency record even when no pending queue remains.
await tx
.select({ id: issueRows.id })
.from(issueRows)
.where(and(eq(issueRows.id, issue.id), eq(issueRows.companyId, issue.companyId)))
.for("update");
const retryWake = await tx
.select()
.from(agentWakeupRequests)
.where(and(
eq(agentWakeupRequests.id, req.body.queueId),
eq(agentWakeupRequests.companyId, issue.companyId),
issue.assigneeAgentId
? eq(agentWakeupRequests.agentId, issue.assigneeAgentId)
: undefined,
))
.for("update")
.limit(1)
.then((rows) => rows[0] ?? null);
const retryRun = retryWake && readObject(retryWake.payload).issueId === issue.id
? await tx
.select()
.from(heartbeatRuns)
.where(and(
eq(heartbeatRuns.id, req.body.targetRunId),
eq(heartbeatRuns.companyId, issue.companyId),
eq(heartbeatRuns.agentId, retryWake.agentId),
))
.for("update")
.limit(1)
.then((rows) => rows[0] ?? null)
: null;
const retryRunContext = readObject(retryRun?.contextSnapshot);
const retryRunResult = readObject(retryRun?.resultJson);
const retryAcknowledgements = readObject(
retryRunResult.queuedSteeringAcknowledgements,
);
const retryAcknowledgement = readObject(retryAcknowledgements[commentId]);
if (
retryRun
&& (retryRunContext.issueId === issue.id || retryRunContext.taskId === issue.id)
&& retryAcknowledgement.status === "acknowledged"
&& retryAcknowledgement.queueId === req.body.queueId
) {
duplicate = true;
acknowledgedTurnId = typeof retryAcknowledgement.turnId === "string"
? retryAcknowledgement.turnId
: null;
return buildQueuedCommentQueue({
executor: tx,
issue,
activeRun: retryRun.status === "running" ? retryRun : null,
actor,
});
}
const locked = await lockQueuedCommentState({
tx,
issue,
actor,
queueId: req.body.queueId,
targetRunId: req.body.targetRunId,
});
if (!locked.activeRun) {
throw conflict("The queued message targets a stale run", {
code: "queued_comment_stale_target",
});
}
const runResult = readObject(locked.activeRun.resultJson);
const acknowledgements = readObject(runResult.queuedSteeringAcknowledgements);
const priorAcknowledgement = readObject(acknowledgements[commentId]);
if (
priorAcknowledgement.status === "acknowledged"
&& priorAcknowledgement.queueId === req.body.queueId
) {
duplicate = true;
acknowledgedTurnId = typeof priorAcknowledgement.turnId === "string"
? priorAcknowledgement.turnId
: null;
return buildQueuedCommentQueue({
executor: tx,
issue,
activeRun: locked.activeRun,
actor,
queueState: locked.queueState,
});
}
assertQueueMutationTarget({
queue: locked.queue,
queueId: req.body.queueId,
revision: req.body.revision,
});
if (locked.queue.protocol !== "paperclip_runner_v1") {
throw conflict("This runner does not support same-turn steering", {
code: "steering_unsupported",
});
}
const entry = locked.queue.entries.find((candidate) => candidate.comment.id === commentId);
if (!entry) {
throw conflict("The queued message is no longer pending", {
code: "queued_comment_not_pending",
});
}
const acknowledgement = await steerNativeSession({
runId: locked.activeRun.id,
message: entry.comment.body,
correlationId: commentId,
});
acknowledgedTurnId = acknowledgement.turnId;
const remainingIds = locked.queue.entries
.map((candidate) => candidate.comment.id)
.filter((candidateId) => candidateId !== commentId);
const now = new Date();
const nextWake = remainingIds.length === 0
? await tx
.update(agentWakeupRequests)
.set({ status: "cancelled", finishedAt: now, updatedAt: now })
.where(eq(agentWakeupRequests.id, locked.wake.id))
.returning()
.then(() => null)
: await tx
.update(agentWakeupRequests)
.set({
payload: withQueuedCommentIdsInWakePayload(locked.wake.payload, remainingIds),
updatedAt: now,
})
.where(eq(agentWakeupRequests.id, locked.wake.id))
.returning()
.then((rows) => rows[0] ?? locked.wake);
await tx
.update(heartbeatRuns)
.set({
resultJson: {
...runResult,
queuedSteeringAcknowledgements: {
...acknowledgements,
[commentId]: {
status: "acknowledged",
queueId: req.body.queueId,
turnId: acknowledgement.turnId,
acknowledgedAt: now.toISOString(),
},
},
},
updatedAt: now,
})
.where(eq(heartbeatRuns.id, locked.activeRun.id));
return buildQueuedCommentQueue({
executor: tx,
issue,
activeRun: locked.activeRun,
actor,
queueState: nextWake
? { wake: nextWake, state: "deferred", queueRun: null }
: null,
});
});
} catch (error) {
if (error instanceof NativeSessionSteeringError) {
throw conflict(error.message, { code: error.code, retryable: true });
}
throw error;
}
await logActivity(db, {
companyId: issue.companyId,
actorType: actor.actorType,
actorId: actor.actorId,
agentId: actor.agentId,
runId: actor.runId,
agentApiKeyId: actor.agentApiKeyId,
action: "issue.queued_comment_steered",
entityType: "issue",
entityId: issue.id,
details: {
commentId,
targetRunId: req.body.targetRunId,
turnId: acknowledgedTurnId,
duplicate,
},
});
res.json(await runRedactions.redactForIssue(issue.companyId, issue.id, queue));
},
);
router.delete(
"/issues/:id/queued-comments/:commentId",
+16
View File
@@ -4702,6 +4702,22 @@ registry.registerPath({
responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized, 404: r.notFound, 409: r.conflict },
});
registry.registerPath({
method: "post",
path: "/api/issues/{id}/queued-comments/{commentId}/steer",
tags: ["issues"],
summary: "Steer a queued issue comment into the active native run",
request: {
params: z.object({ id: z.string(), commentId: z.string() }),
body: jsonBody(z.object({
queueId: z.string().min(1),
revision: z.string().min(1),
targetRunId: z.string().min(1),
})),
},
responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized, 403: r.forbidden, 404: r.notFound, 409: r.conflict },
});
registry.registerPath({
method: "post",
path: "/api/heartbeat-runs/{runId}/runtime-requests/{requestId}/resolve",