fix: preserve task continuation and question delivery

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-20 20:13:36 -05:00
1 parent 8b4ab65b1c
commit fc90c686e5
18 files changed
+449 -11

No files matched your search

+12
View File
@@ -141,6 +141,18 @@ records are evidence and do not grant permission to resolve another review.
### Answered Slack conversations
An accepted scheduled-monitor yield remains unfinished task work. Its committed
summary is visible as a continuation explanation on the board and its authorized
chat conversation. A run-owned question or confirmation keeps the response slot;
the explanation must not duplicate that interaction. Presentation recovery can
materialize the saved summary once after workspace finalization, without another
provider turn. Run success alone cannot produce a task-completion claim.
Monitor replies inherit a conversation only through an admitted monitor wake
and the matching server-issued scheduling receipt on the source run. Company,
task, agent, requester, schedule identity, and current delivery authorization
remain enforced. Provider-supplied source IDs do not establish that lineage.
A successful Slack turn with a published final reply and no remaining execution
or decision path settles to `chat_conversations.state = waiting` and issue
`in_review`. This is a server-owned passive conversation state, displayed as
@@ -0,0 +1,63 @@
# Dependable everyday work
## Acceptance
Codex and Claude must finish ordinary engineering tasks through both the board
and Slack, including authorized repository setup, automatic continuation,
questions, restart recovery, follow-ups, Stop, and accessible result delivery.
Normal rollout continues. Each core cell needs three fresh successful attempts
at a recorded candidate revision and deployed staging proof; failed attempts
remain evidence. A provider success or a green task badge is insufficient.
## Delivery and ownership
The implementation lead owns the composed journey. The first repair slice owns
task-outcome presentation, monitor publication provenance, and complete question
cards. The second slice owns the six Product E2E journeys and explicit daily
selection, reusing the existing campaign accounting and reporting pipeline.
| Boundary | Required evidence | State |
| --- | --- | --- |
| Yielded monitor presentation | Explanation retained; no false completion or duplicate question | Repair and focused checks pass |
| Monitor conversation lineage | Exact scheduling receipt; current authorization; no unrelated channel disclosure | Repair and database checks pass |
| Free-text question card | Prompt, context, and Respond visible; answers stay private | Repair and database checks pass |
| Repository preparation | Authorized context; automatic successor; retained workspace and tools | Inspecting existing repairs |
| Finalization recovery | Saved result survives; zero repeated completed provider turns | Saved-summary repair and database checks pass |
| Migration and follow-ups | Configuration retained and subsequent work delivered | Qualification pending |
## Qualification matrix
Six journeys, each on Codex and Claude through Slack and the board:
1. Repository task with independently verified accessible output.
2. Projectless task, repository preparation, automatic continuation, output.
3. Legacy-to-native conversion retaining configuration, followed by useful work.
4. Necessary free-text question, answer, and exactly one continuation.
5. Restart during finalization, recovery, and exactly one final delivery.
6. Active follow-up, Stop, later resumption, and retained context.
Use production instructions, recorded runtime configurations, immutable remote
images, isolated tasks, and a dedicated test channel. Negative checks cover
revocation, forged lineage, duplicate wakes, stale settings, and restricted
workspaces. Keep source, runner, model, settings, evidence, timing, cost, and
attempt identity with every result. Never infer historical qualification from
the presence of a fixture.
## Baseline
Source base: `8b4ab65b1c3d863d282752e83307943720a5c85c`.
Private incident evidence stays outside the public repository. Serving revision,
runner artifact, and historical successful deployment still require independent
capture. Workspace and lifecycle changes identify changed execution paths, not
an established single introducing cause. The earlier staging settings regression
already has a merged repair; verify it rather than duplicating that change.
## Verification record
The original source fails four composed database regressions: monitor origin,
yielded completion, question body, and saved-summary recovery. The initial
response-resolver regression also fails independently on the original source.
The repaired six-file cohort passes 148 checks against real disposable PostgreSQL,
including existing monitor scheduling, revocation, and recovery boundaries.
Workspace typecheck passes. Full build, full suite, review, and live qualification
remain separate gates. No deployed acceptance is claimed here.
+7
View File
@@ -95,6 +95,13 @@ remaining operations before qualifying the whole capability.
Gaps: Normal agent replies work through server delivery. Explicit endpoint publication
APIs are board management paths.
Monitor publication lineage uses the admitted wake and its source run's matching
`set_task_monitor` receipt, including company, task, immutable agent, requester,
and scheduled time. A source ID in context cannot grant publication. The normal
delivery worker rechecks current endpoint and sender authority. Database fixtures
exercise lineage and finalization recovery; live Slack qualification remains
separate.
Publication, provider-reply, and form-open 403/permission errors fail the rejected
operation without removing unrelated assigned channel tools. Confirmed
authentication failures still put the endpoint in attention, which blocks tool
@@ -46,6 +46,10 @@ Verification: `not-run` — Create complete canonical mixed form; same idempoten
Gaps: Legacy wrappers and native dedicated tool have different fields; source availability does not establish every harness.
Provider question cards retain the creation summary and each question's prompt
and help text when the answer needs a modal. This changes presentation only;
answer authorization and the private terminal receipt remain unchanged.
### `answer`
API: `present` — POST /api/issues/{id}/interactions/{interactionId}/respond.
+7
View File
@@ -47,6 +47,13 @@ Setup is covered in [connection setup](./connection-setup.md).
Automated: [channel integration](../../server/src/__tests__/chat-channels.integration.test.ts) uses provider fixtures; it does not prove a live Slack, Discord, Teams, Telegram, GitHub, email, or Photon delivery.
Monitor continuation regression: [database checks](../../server/src/services/chat-monitor-continuation.test.ts)
verify the exact originating conversation, rejected forged/mismatched receipts,
one working publication, no completion for a yielded run, and saved-response
recovery. In a disposable live thread, request work that needs deferred repository
preparation. Verify a useful continuation explanation, automatic resumption, and
the final result in that same thread. A successful provider turn alone is insufficient.
GitHub review comment regression (2026-10-19): repeated PR review requests reuse
one main App comment, including across restart and reconnect. Test a leading
request for an external bot (`@greptile review`) and verify it creates no wake or
@@ -187,6 +187,10 @@ Automated: [chat question forms](../../server/src/services/chat-question-forms.t
and [interaction publications](../../server/src/__tests__/chat-interaction-publications.test.ts)
cover serialization and delivery logic with fixtures.
Free-text and multi-question cards must show their title, question prompts, help
text, and relevant context beside Respond. Settled cards must not expose saved
free-text answers. The interaction publication database suite covers both states.
Manual: use a disposable provider conversation, answer as the linked sender,
and check the task's persisted result and one continuation. Repeat from an
unlinked or wrong sender and with an expired card. A browser answer does not
@@ -59871,7 +59871,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
{
threadId: dm.thread.id,
messageId: "outbound-2",
text: "Maya completed this turn.",
text: "Maya has no final response to show yet. Open the task in Paperclip for details.",
},
]);
expect(
@@ -60014,7 +60014,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
messageId: "outbound-2",
text: publishesComment
? `Explicit Telegram reply via ${authorizationReason}`
: "Maya completed this turn.",
: "Maya has no final response to show yet. Open the task in Paperclip for details.",
},
]);
await expect(
@@ -230,6 +230,35 @@ describeEmbeddedPostgres(
};
}
it("shows the free-text question and context alongside Respond", async () => {
const fixture = await seedBoundIssue(["slack"]);
const interaction = await issueThreadInteractionService(db).create(
{ id: fixture.issueId, companyId: fixture.companyId },
{
kind: "ask_user_questions",
title: "Identify the failed change",
summary: "The general migration checks passed. I need one detail to reproduce your specific failure.",
payload: {
version: 1,
title: "Identify the failed change",
questions: [{
id: "agent", prompt: "Which agent did you try to change?",
helpText: "Send the agent link and the error text, if available.",
selectionMode: "single", allowOther: true,
options: [{ id: "text", label: "Type an answer", freeText: true }],
}],
},
},
{ agentId: fixture.agentId },
);
const [publication] = await publicationsForInteraction(fixture.companyId, interaction.id);
expect(publication.payload.card).toMatchObject({
title: "Identify the failed change",
body: "The general migration checks passed. I need one detail to reproduce your specific failure.\n\nWhich agent did you try to change?\n\nSend the agent link and the error text, if available.",
actions: [expect.objectContaining({ label: "Respond", type: "callback" })],
});
});
it("publishes prompts only for the endpoint's verified immutable agent", async () => {
const fixture = await seedBoundIssue();
const service = issueThreadInteractionService(db);
@@ -24,6 +24,43 @@ describe("model-facing run errors", () => {
});
});
describe("scheduled continuation presentation", () => {
const resultJson = {
finalizationPhase: "committed",
finalizationReasonCode: "scheduled_monitor_waiting",
nativeResult: {
schema: "paperclip.run_result.v1",
reportedWorkDisposition: "yielded",
summary: "QA is still pending. The repository is being prepared and I will continue automatically.",
continuation: { kind: "monitor", summary: "Continue with the prepared repository.", idempotencyKey: "repository-ready" },
},
};
it.each([false, true])("retains the accepted continuation explanation on board and chat (%s)", (external) => {
expect(resolveHeartbeatRunResponse({
resultJson,
preferFinalResponseOverExistingComment: external,
finalAgentMessage: { text: "The task is complete.", sourceEventId: "provider-final", channel: "final" },
})).toMatchObject({
text: resultJson.nativeResult.summary,
decision: { chosenSource: "semantic_result_summary", commentAction: "create", reasonCodes: ["accepted_scheduled_monitor_summary"] },
});
});
it("withholds continuation claims until finalization commits", () => {
expect(resolveHeartbeatRunResponse({ resultJson: { ...resultJson, finalizationPhase: "retryable_failure" } }).text).toBeNull();
});
it("reuses its saved continuation comment after recovery", () => {
expect(resolveHeartbeatRunResponse({ resultJson, existingComment: { id: "saved-continuation", body: resultJson.nativeResult.summary } }))
.toMatchObject({ text: resultJson.nativeResult.summary, decision: { commentAction: "reuse", commentId: "saved-continuation" } });
});
it("lets the run's question own its response", () => {
expect(resolveHeartbeatRunResponse({ resultJson, interactionOwnsResponse: true }).text).toBeNull();
});
});
describe("selectHeartbeatRunFinalAgentMessage", () => {
const substantive = {
seq: 80,
@@ -193,8 +193,10 @@ function textForQuestionInteraction(
interaction.payload.title ?? interaction.title ?? "Input needed",
"",
];
if (interaction.summary) lines.push(interaction.summary, "");
for (const question of interaction.payload.questions) {
lines.push(question.prompt);
if (question.helpText) lines.push(question.helpText);
for (const option of question.options) lines.push(`- ${option.label}`);
lines.push("");
}
@@ -401,7 +403,12 @@ export async function enqueueIssueInteractionChatPublications(
: "Response needed in Paperclip",
body:
interaction.kind === "ask_user_questions"
? (question?.helpText ?? undefined)
? [...new Set([
interaction.summary,
...(question
? [question.helpText]
: interaction.payload.questions.flatMap(item => [item.prompt, item.helpText])),
].filter((value): value is string => Boolean(value?.trim())))].join("\n\n") || undefined
: interaction.kind === "request_confirmation"
? (interaction.payload.detailsMarkdown ?? undefined)
: "Open the task in Paperclip to review and respond.",
@@ -0,0 +1,144 @@
import { randomUUID } from "node:crypto";
import { and, eq } from "drizzle-orm";
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import {
agents, agentWakeupRequests, authUsers, chatConversations, chatEndpoints, chatMessageLinks,
chatPublications, companies, completionContracts, createDb, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults,
toolApplications, toolConnections,
} from "@paperclipai/db";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js";
import { enqueueChatRunMilestones, resolveChatRunPresentationAuthorizationReason } from "./chat-run-publications.js";
import { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON } from "./heartbeat-run-summary.js";
import { resolveChatOriginPublicationBindings } from "./issues.js";
import { NativeRunCoordinatorStore } from "./native-runtime/native-run-coordinator-store.js";
import { finalizeNativeRun, repairCommittedNativeChatResponse } from "./native-runtime/native-run-finalizer.js";
const support = await getEmbeddedPostgresTestSupport();
if (!support.supported) console.warn(`Monitor publication database unavailable: ${support.reason}`);
(support.supported ? describe : describe.skip)("monitor conversation continuation", () => {
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: ReturnType<typeof createDb>;
beforeAll(async () => {
temporary = await startEmbeddedPostgresTestDatabase("paperclip-monitor-publications-");
db = createDb(temporary.connectionString);
await db.insert(authUsers).values({ id: "requester", name: "Requester", email: "requester@example.test", createdAt: new Date(), updatedAt: new Date() });
});
afterAll(async () => { await temporary?.cleanup(); });
async function fixture() {
const [company] = await db.insert(companies).values({ name: "Continuation", issuePrefix: randomUUID().slice(0, 8) }).returning();
const [agent] = await db.insert(agents).values({ companyId: company.id, name: "Worker", adapterType: "paperclip_runner" }).returning();
const [issue] = await db.insert(issues).values({ companyId: company.id, title: "Repository QA", status: "in_progress", assigneeAgentId: agent.id }).returning();
const [application] = await db.insert(toolApplications).values({ companyId: company.id, applicationKey: randomUUID(), name: "Slack", type: "chat" }).returning();
const [connection] = await db.insert(toolConnections).values({ companyId: company.id, applicationId: application.id, name: "Slack", uid: randomUUID(), connectionPurpose: "channel", transport: "chat_sdk" }).returning();
const [endpoint] = await db.insert(chatEndpoints).values({ companyId: company.id, connectionId: connection.id, provider: "slack", publicId: randomUUID(), assignedAgentId: agent.id, status: "active" }).returning();
const [conversation] = await db.insert(chatConversations).values({ companyId: company.id, endpointId: endpoint.id, issueId: issue.id, externalConversationId: "test-channel", externalThreadId: randomUUID(), externalLabel: "QA" }).returning();
const [comment] = await db.insert(issueComments).values({ companyId: company.id, issueId: issue.id, authorType: "user", authorUserId: "requester", body: "QA the runner migration" }).returning();
await db.insert(chatMessageLinks).values({ companyId: company.id, endpointId: endpoint.id, conversationId: conversation.id, commentId: comment.id, providerMessageId: randomUUID(), direction: "inbound" });
const nextCheckAt = new Date(Date.now() + 60_000).toISOString();
const receipt = { operationId: "set_task_monitor", result: { taskId: issue.id, monitor: { status: "scheduled", nextCheckAt } } };
const [source] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, nativeIssueId: issue.id, runtimeMode: "native", status: "succeeded", responsibleUserId: "requester",
contextSnapshot: { source: "chat:slack", issueId: issue.id, endpointId: endpoint.id, wakeCommentId: comment.id },
resultJson: { semanticToolReceipts: { schedule: receipt } },
}).returning();
const [successor] = await db.insert(heartbeatRuns).values({ companyId: company.id, agentId: agent.id, nativeIssueId: issue.id, runtimeMode: "native", status: "running", responsibleUserId: "requester",
contextSnapshot: { source: "issue.monitor", issueId: issue.id, wakeReason: "issue_monitor_due", nextCheckAt },
}).returning();
const [wake] = await db.insert(agentWakeupRequests).values({ companyId: company.id, agentId: agent.id, source: "automation", reason: "issue_monitor_due", status: "claimed", runId: successor.id,
idempotencyKey: `issue-monitor:${issue.id}:${nextCheckAt}`, payload: { issueId: issue.id, nextCheckAt },
}).returning();
const bindings = (runId = successor.id) => resolveChatOriginPublicationBindings(db, company.id, issue.id, runId);
return { company, agent, issue, endpoint, conversation, source, successor, wake, receipt, bindings };
}
it("retains the exact originating conversation through the admitted monitor receipt", async () => {
const f = await fixture();
expect(await f.bindings(f.source.id)).toHaveLength(1);
expect(await f.bindings()).toEqual(await f.bindings(f.source.id));
expect(await resolveChatRunPresentationAuthorizationReason(db, { companyId: f.company.id, issueId: f.issue.id, runId: f.successor.id }))
.toBe(CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON);
expect(await enqueueChatRunMilestones(db)).toBe(1);
expect(await enqueueChatRunMilestones(db)).toBe(0);
const publications = await db.select().from(chatPublications).where(eq(chatPublications.issueId, f.issue.id));
expect(publications).toHaveLength(1);
expect(publications[0]).toMatchObject({ conversationId: f.conversation.id, payload: { progressState: "working" } });
});
it.each(["receipt", "wake", "requester", "agent", "schedule", "task"])("rejects broken %s lineage even with a claimed source id", async fault => {
const f = await fixture();
await db.update(heartbeatRuns).set({ contextSnapshot: { ...f.successor.contextSnapshot, sourceRunId: f.source.id } }).where(eq(heartbeatRuns.id, f.successor.id));
if (fault === "receipt") await db.update(heartbeatRuns).set({ resultJson: {} }).where(eq(heartbeatRuns.id, f.source.id));
if (fault === "wake") await db.update(agentWakeupRequests).set({ runId: null }).where(eq(agentWakeupRequests.id, f.wake.id));
if (fault === "requester") await db.update(heartbeatRuns).set({ responsibleUserId: "another-person" }).where(eq(heartbeatRuns.id, f.successor.id));
if (fault === "agent") {
const [other] = await db.insert(agents).values({ companyId: f.company.id, name: "Other", adapterType: "paperclip_runner" }).returning();
await db.update(heartbeatRuns).set({ agentId: other.id }).where(eq(heartbeatRuns.id, f.source.id));
}
if (fault === "schedule") await db.update(agentWakeupRequests).set({ idempotencyKey: randomUUID() }).where(eq(agentWakeupRequests.id, f.wake.id));
if (fault === "task") {
const [other] = await db.insert(issues).values({ companyId: f.company.id, title: "Private task" }).returning();
await db.update(heartbeatRuns).set({ resultJson: { semanticToolReceipts: { schedule: {
...f.receipt, result: { ...f.receipt.result, taskId: other.id },
} } } }).where(eq(heartbeatRuns.id, f.source.id));
}
expect(await f.bindings()).toEqual([]);
});
it("does not turn a yielded successful provider run into a completion", async () => {
const f = await fixture();
await db.update(heartbeatRuns).set({ status: "succeeded", resultJson: {
presentationDecision: { commentAction: "none" },
nativeResult: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", continuation: { kind: "monitor" } },
} }).where(eq(heartbeatRuns.id, f.source.id));
await enqueueChatRunMilestones(db);
expect(await db.select().from(chatPublications).where(and(eq(chatPublications.issueId, f.issue.id), eq(chatPublications.idempotencyKey, `run:${f.source.id}:completed:${f.endpoint.id}`)))).toEqual([]);
});
it("recovers the accepted continuation explanation after interrupted finalization without another provider result", async () => {
const f = await fixture();
const [contract] = await db.insert(completionContracts).values({ companyId: f.company.id, issueId: f.issue.id,
revision: 1, schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard",
completionAuthority: "server_arbiter", incompleteCriteriaPolicy: "preserve_non_terminal",
contractJson: { revision: "test", objective: "Verify repository migration", criteria: [] }, canonicalSha256: "monitor-contract",
createdByActorType: "system", createdByActorId: "test",
}).returning();
const sessionId = randomUUID(), runnerId = randomUUID();
await db.update(heartbeatRuns).set({ status: "running", nativeSessionId: sessionId, runnerInstanceId: runnerId,
completionContractId: contract.id, completionContractSha256: contract.canonicalSha256,
}).where(eq(heartbeatRuns.id, f.source.id));
await db.update(heartbeatRuns).set({ status: "cancelled", finishedAt: new Date() }).where(eq(heartbeatRuns.id, f.successor.id));
await db.update(issues).set({ executionRunId: f.source.id, monitorNextCheckAt: new Date(f.receipt.result.monitor.nextCheckAt),
monitorScheduledBy: "assignee", executionPolicy: { monitor: { nextCheckAt: f.receipt.result.monitor.nextCheckAt, notes: "Continue QA" } },
}).where(eq(issues.id, f.issue.id));
const port = new NativeRunCoordinatorStore(db, { companyId: f.company.id, issueId: f.issue.id,
runId: f.source.id, agentId: f.agent.id, normalizedSessionId: sessionId, completionContractId: contract.id,
completionContractSha256: contract.canonicalSha256, runnerSourceInstanceId: runnerId,
completionContractRevision: "test", completionContractCriterionIds: [],
});
const summary = "The repository is being prepared. I will continue QA automatically.";
await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", summary,
completionClaim: { contractRevision: "test", objectiveSatisfied: false, criteria: [], remainingWork: [{ description: "Complete QA", blocksCompletion: true }] },
continuation: { kind: "monitor", summary: "Continue QA", idempotencyKey: "prepare-source" },
evidence: [], verification: [], attentionRequests: [], artifacts: [],
}, terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "yielded" }, turnId: "work-turn" });
await finalizeNativeRun({ db, runId: f.source.id, workspaceFinalizeStatus: "failed", projectRunStatus: true });
const repair = () => repairCommittedNativeChatResponse(db, { companyId: f.company.id, issueId: f.issue.id, runId: f.source.id });
expect(await repair()).toBe(false);
await finalizeNativeRun({ db, runId: f.source.id, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.source.id));
expect(coordinator).toMatchObject({ phase: "committed" });
const [settledRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.source.id));
expect(settledRun).toMatchObject({ status: "succeeded", resultJson: { finalizationPhase: "committed" } });
const priorComments = await db.select({ body: issueComments.body }).from(issueComments).where(eq(issueComments.createdByRunId, f.source.id));
expect(priorComments).toEqual([{ body: summary }]);
expect(await repair()).toBe(false);
expect(await repair()).toBe(false);
const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, f.source.id));
expect(comments.map(comment => comment.body)).toEqual([summary]);
expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.source.id))).toHaveLength(1);
expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.company.id))).toHaveLength(1);
const publications = await db.select().from(chatPublications).where(eq(chatPublications.issueId, f.issue.id));
expect(publications).toHaveLength(1);
expect(publications[0]).toMatchObject({ conversationId: f.conversation.id, payload: { text: summary } });
});
});
@@ -163,14 +163,14 @@ describe("chat run milestone projection", () => {
).toBe("Maya stopped at your request.");
});
it("projects a successful run without an explicit reply as a generic completion", () => {
it("does not claim task completion when a successful run has no response", () => {
expect(
safeMilestoneText({
agentName: "Maya",
milestone: "completed",
issueId: "issue-1",
}),
).toBe("Maya completed this turn.");
).toBe("Maya has no final response to show yet. Open the task in Paperclip for details.");
});
});
+24 -5
View File
@@ -35,6 +35,7 @@ import { safeChatTaskUrl } from "./chat-task-url.js";
import { hasChatRunOwnedProviderInteraction } from "./chat-interaction-arbitration.js";
import { CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON } from "./heartbeat-run-summary.js";
import { resolveChatOriginPublicationBindings } from "./issues.js";
import { boundRunIssueIdSql } from "./run-task-binding.js";
import { authorizeNativeChatReviewPresentation } from "./native-runtime/native-chat-review-presentation.js";
import {
SAFE_NATIVE_CHAT_PROGRESS_EVENT_TYPES,
@@ -250,13 +251,13 @@ export function safeMilestoneText(input: {
}): string {
if (input.milestone === "queued") return `${input.agentName} is queued.`;
if (input.milestone === "working") return `${input.agentName} is working…`;
if (input.milestone === "completed")
return `${input.agentName} completed this turn.`;
if (input.errorCode === "slack_session_stopped")
return `${input.agentName} stopped at your request.`;
const taskUrl = safeChatTaskUrl(input.publicBaseUrl, input.issueId);
const recovery =
input.milestone === "waiting_for_input"
input.milestone === "completed"
? `${input.agentName} has no final response to show yet.`
: input.milestone === "waiting_for_input"
? `${input.agentName} needs a Paperclip admin to safely recover this turn before more work can start.`
: input.errorCode === "low_trust_isolation_unavailable"
? `${input.agentName} couldn't safely start this turn because this task was started for an unlinked external guest and isolated guest execution isn't available. Ask a Paperclip admin to create a private identity link for this account or enable isolated guest execution, then start a new task.`
@@ -291,7 +292,7 @@ async function enqueueSafeNativeChatProgress(
heartbeatRunEvents,
"safe_native_progress_later_event",
);
const issueIdFromContext = sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`;
const issueIdFromContext = boundRunIssueIdSql();
let inserted = 0;
let cursor: {
eventCreatedAt: Date;
@@ -667,7 +668,7 @@ export async function enqueueChatRunMilestones(
): Promise<number> {
const since = input.since ?? new Date(Date.now() - 24 * 60 * 60_000);
const limit = Math.max(1, Math.min(input.limit ?? 200, 1_000));
const issueIdFromContext = sql<string>`${heartbeatRuns.contextSnapshot} ->> 'issueId'`;
const issueIdFromContext = boundRunIssueIdSql();
const explicitlyAuthoredCommentReason = sql<boolean>`(
${issueComments.metadata} ->> 'authorizationReason' = 'paperclip_runner_protocol'
or left(coalesce(${issueComments.metadata} ->> 'authorizationReason', ''), 6) = 'allow_'
@@ -736,6 +737,17 @@ export async function enqueueChatRunMilestones(
)
)
)`;
const hasMonitorContinuation = sql<boolean>`(
${heartbeatRuns.contextSnapshot}->>'source' = 'issue.monitor' and exists (
select 1 from agent_wakeup_requests monitor_wake
where monitor_wake.company_id = ${heartbeatRuns.companyId}
and monitor_wake.run_id = ${heartbeatRuns.id}
and monitor_wake.agent_id = ${heartbeatRuns.agentId}
and monitor_wake.reason = 'issue_monitor_due'
and monitor_wake.status <> 'skipped'
and monitor_wake.payload->>'issueId' = ${issueIdFromContext}
)
)`;
let inserted = 0;
let cursor: {
updatedAt: Date;
@@ -822,6 +834,7 @@ export async function enqueueChatRunMilestones(
hasDirectInteractionContinuation,
),
hasQuestionContinuationTarget,
hasMonitorContinuation,
),
// Heartbeat marks a run succeeded before the presentation resolver
// finishes. Waiting for its durable decision prevents a generic
@@ -830,6 +843,11 @@ export async function enqueueChatRunMilestones(
${heartbeatRuns.status} <> 'succeeded'
or ${heartbeatRuns.resultJson} -> 'presentationDecision' is not null
)`,
sql`(${heartbeatRuns.status} <> 'succeeded' or coalesce(
${heartbeatRuns.resultJson}->'nativeResult'->>'reportedWorkDisposition',
${heartbeatRuns.resultJson}->'acceptedResult'->>'reportedWorkDisposition',
${heartbeatRuns.resultJson}->'semanticResult'->'result'->>'reportedWorkDisposition', ''
) <> 'yielded')`,
// A successful or interrupted run's selected final remains its
// authoritative response. Filter it before LIMIT so a page of settled
// runs cannot starve later milestones.
@@ -865,6 +883,7 @@ export async function enqueueChatRunMilestones(
),
),
or(
hasMonitorContinuation,
sql`exists (
select 1
from ${chatMessageLinks}
+36 -1
View File
@@ -130,7 +130,7 @@ const NARRATION_OPENERS =
export const LEGACY_WITHHELD_RUN_COMMENT =
"Run completed. Agent did not post a summary comment this run (transcript withheld — see run log).";
export const RUN_PRESENTATION_RESOLVER_VERSION = "1";
export const RUN_PRESENTATION_RESOLVER_VERSION = "2";
export const CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON =
"allow_chat_run_presentation";
@@ -204,6 +204,7 @@ export function isExternalChatPresentationContext(
// durable source-run or mirrored-comment proof authorizes an external reply.
if (
source === "tool_action_review" ||
source === "issue.monitor" ||
source.startsWith("issue.comment") ||
source === "issue.update"
)
@@ -284,6 +285,22 @@ function hasYieldedSemanticResult(resultJson: Record<string, unknown>) {
);
}
export function readAcceptedMonitorContinuationSummary(
resultJson: Record<string, unknown> | null | undefined,
): string | null {
const result = record(resultJson);
if (result.finalizationPhase !== "committed" ||
!["scheduled_monitor_waiting", "scheduled_monitor_dispatched"].includes(String(result.finalizationReasonCode))) return null;
for (const candidate of semanticResultCandidates(result)) {
const continuation = record(candidate.continuation);
if (candidate.schema === "paperclip.run_result.v1" && candidate.reportedWorkDisposition === "yielded" &&
continuation.kind === "monitor" && readCommentText(continuation.idempotencyKey)) {
return readCommentText(candidate.summary);
}
}
return null;
}
function readAcceptedExternalChatResponseWakeSummary(
resultJson: Record<string, unknown>,
reviewPresentationAuthorized = false,
@@ -379,6 +396,7 @@ export function resolveHeartbeatRunResponse(input: {
/** Server-owned conversation finalization, after normal governance checks. */
conversationTurnFinished?: boolean;
existingComment?: { id: string; body?: string | null } | null;
interactionOwnsResponse?: boolean;
preferFinalResponseOverExistingComment?: boolean;
externalChatResponseWakeSummaryAuthorized?: boolean;
/** Server-only proof of the exact accepted response after durable status
@@ -393,6 +411,23 @@ export function resolveHeartbeatRunResponse(input: {
} | null;
}): ResolvedHeartbeatRunResponse {
const resultJson = record(input.resultJson);
const monitorSummary = readAcceptedMonitorContinuationSummary(resultJson);
if (monitorSummary) {
if (input.interactionOwnsResponse) return {
text: null,
decision: decision("none", { commentAction: "none", reasonCodes: ["interaction_owns_response"] }),
};
const existing = input.existingComment;
const reuse = existing?.body === monitorSummary;
return {
text: monitorSummary,
decision: decision("semantic_result_summary", {
commentAction: reuse ? "reuse" : "create",
...(reuse ? { commentId: existing.id } : {}),
reasonCodes: ["accepted_scheduled_monitor_summary"],
}),
};
}
const finalAgentText = readCommentText(input.finalAgentMessage?.text);
const explicitProviderFinal = input.finalAgentMessage?.channel === "final";
const compatibleTerminalAssistant =
@@ -27,6 +27,8 @@ import { logActivity } from "../activity-log.js";
import { isUnresolvedWorkspaceBaseRefError, readUnresolvedWorkspaceBaseRefDiagnostic, type UnresolvedWorkspaceBaseRefError } from "../workspace-runtime.js";
import { runnerGoalService } from "../runner-goals.js";
import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js";
import { hasChatRunOwnedProviderInteraction } from "../chat-interaction-arbitration.js";
import { readAcceptedMonitorContinuationSummary } from "../heartbeat-run-summary.js";
import { SANDBOX_PROVIDER_PLUGIN_NOT_READY_REASON } from "../recovery/stranded-notice.js";
import { MAX_TURN_CONTINUATION_RETRY_REASON } from "../../modules/run-dispatch/index.js";
import { redactCurrentUserText } from "../../log-redaction.js";
@@ -789,6 +791,7 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
));
const externalChatPresentationCandidate =
isExternalChatPresentationContext(livenessRun.contextSnapshot) ||
parseObject(livenessRun.contextSnapshot).source === "issue.monitor" ||
parseObject(livenessRun.contextSnapshot).source === "tool_action_review" ||
String(parseObject(livenessRun.contextSnapshot).source ?? "").startsWith("issue.comment") ||
parseObject(livenessRun.contextSnapshot).source === "issue.update";
@@ -806,6 +809,8 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC
);
const resolved = resolveHeartbeatRunResponse({
resultJson: persistedResultJson,
interactionOwnsResponse: Boolean(issueId && readAcceptedMonitorContinuationSummary(persistedResultJson) &&
await hasChatRunOwnedProviderInteraction(db, { companyId: livenessRun.companyId, issueId, runId: livenessRun.id })),
conversationTurnFinished: isConversation(issueContext) && livenessRun.status === "succeeded"
&& persistedResultJson?.finalizationReasonCode === "conversation_turn_finished",
existingComment: existingRunComment,
+7
View File
@@ -1,4 +1,5 @@
import { measureRunTransaction } from "./execution-transaction-timing.js";
import { readNativeMonitorSource } from "./native-runtime/native-monitor-delivery.js";
import { assertNoCaseHandoffMutation, cancelCaseHandoffsForIssue, completeTerminalCaseHandoffsForIssue } from "./heartbeat/case-handoff-lifecycle.js";
import { measureRunPhase } from "./execution-preparation-timing.js";
import { automaticReadRecoveryCleanupCondition, directHumanCommentCondition, humanCreatedIssueCondition, pendingUserRequestCondition, touchedByUserCondition } from "./issue-participation.js";
@@ -1387,6 +1388,12 @@ export async function resolveChatOriginPublicationBindings(
contextSnapshot = snapshot;
break;
}
if (contextSource === "issue.monitor") {
const sourceRunId = await readNativeMonitorSource(dbOrTx, { companyId, issueId, runId: originRunId });
if (!sourceRunId) return [];
originRunId = sourceRunId;
continue;
}
if (
contextSource !== "issue.interaction.respond" &&
contextSource !== "issue.interaction.accept" &&
@@ -1,5 +1,7 @@
import { agentWakeupRequests, heartbeatRuns, type Db } from "@paperclipai/db";
import { and, desc, eq, inArray, ne, sql } from "drizzle-orm";
import { alias } from "drizzle-orm/pg-core";
import { boundRunIssueIdSql } from "../run-task-binding.js";
function record(value: unknown): Record<string, unknown> {
return value !== null && typeof value === "object" && !Array.isArray(value)
@@ -8,6 +10,41 @@ function record(value: unknown): Record<string, unknown> {
export type NativeMonitorDelivery = { wakeId: string; successorRunId: string };
export async function readNativeMonitorSource(
db: Pick<Db, "select">,
binding: { companyId: string; issueId: string; runId: string },
): Promise<string | null> {
const source = alias(heartbeatRuns, "monitor_source");
const sources = await db.select({ runId: source.id }).from(heartbeatRuns)
.innerJoin(agentWakeupRequests, and(
eq(agentWakeupRequests.companyId, binding.companyId),
eq(agentWakeupRequests.runId, heartbeatRuns.id),
eq(agentWakeupRequests.agentId, heartbeatRuns.agentId),
eq(agentWakeupRequests.reason, "issue_monitor_due"),
ne(agentWakeupRequests.status, "skipped"),
sql`${agentWakeupRequests.payload}->>'issueId' = ${binding.issueId}`,
sql`${agentWakeupRequests.idempotencyKey} = 'issue-monitor:' || ${binding.issueId} || ':' || (${agentWakeupRequests.payload}->>'nextCheckAt')`,
))
.innerJoin(source, and(
eq(source.companyId, binding.companyId), eq(source.agentId, heartbeatRuns.agentId),
eq(source.nativeIssueId, binding.issueId), eq(source.runtimeMode, "native"),
ne(source.id, heartbeatRuns.id),
sql`${source.responsibleUserId} is not distinct from ${heartbeatRuns.responsibleUserId}`,
sql`${source.createdAt} <= ${agentWakeupRequests.createdAt}`,
sql`exists (select 1 from jsonb_each(case
when jsonb_typeof(${source.resultJson}->'semanticToolReceipts') = 'object'
then ${source.resultJson}->'semanticToolReceipts' else '{}'::jsonb end) receipt
where receipt.value->>'operationId' = 'set_task_monitor'
and receipt.value->'result'->>'taskId' = ${binding.issueId}
and receipt.value->'result'->'monitor'->>'status' = 'scheduled'
and receipt.value->'result'->'monitor'->>'nextCheckAt' = ${agentWakeupRequests.payload}->>'nextCheckAt')`,
))
.where(and(eq(heartbeatRuns.id, binding.runId), eq(heartbeatRuns.companyId, binding.companyId),
eq(boundRunIssueIdSql(), binding.issueId)))
.groupBy(source.id).limit(2);
return sources.length === 1 ? sources[0]!.runId : null;
}
/** A dispatched timer is historical delivery evidence, never authority to run again. */
export async function readNativeMonitorDelivery(
db: Db,
@@ -68,11 +68,13 @@ import {
CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON,
findHeartbeatRunCompletionComment,
isExternalChatPresentationContext,
readAcceptedMonitorContinuationSummary,
readCompletedAssistantMessageCandidate,
resolveHeartbeatRunResponse,
selectHeartbeatRunFinalAgentMessage,
} from "../heartbeat-run-summary.js";
import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js";
import { hasChatRunOwnedProviderInteraction } from "../chat-interaction-arbitration.js";
import {
authorizeCommittedChatResponse,
CommittedChatResponseAuthorizationError,
@@ -961,6 +963,25 @@ export async function repairCommittedNativeChatResponse(
// presentation contract, including selected attachments. Do not bypass it.
if (record(decision.decisionJson).externalChatReviewPresentation)
return false;
const monitorResult = { nativeResult: result, finalizationPhase: "committed", finalizationReasonCode: decision.reasonCode };
if (readAcceptedMonitorContinuationSummary(monitorResult) && run.status === "succeeded" &&
terminal.runTerminalState === "succeeded" && terminal.turnTerminalState === "completed" &&
terminal.reportedWorkDisposition === "yielded" && record(run.contextSnapshot).skipIssueComment !== true &&
!await hasChatRunOwnedProviderInteraction(tx, input)) {
const authorizationReason = await resolveChatRunPresentationAuthorizationReason(tx, input);
const resolved = resolveHeartbeatRunResponse({ resultJson: monitorResult });
if (!resolved.text) return false;
const comment = await issueService(db).addComment(input.issueId, resolved.text,
{ agentId: run.agentId, runId: run.id }, { authorizationReason, completionReply: true }, tx);
await tx.update(heartbeatRuns).set({
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
presentationDecision: { ...resolved.decision, commentId: comment.id,
reasonCodes: [...resolved.decision.reasonCodes, "committed_monitor_response_recovered"] },
})}::jsonb`, updatedAt: new Date(),
}).where(eq(heartbeatRuns.id, run.id));
agentId = run.agentId;
return true;
}
// Copyback can be owned by reconciliation before the live heartbeat
// reaches its presentation step. Recover a completed internal task's
// exact final reply from the same accepted turn, without rerunning work