Files
PaperClipAI/ui/src/api/heartbeats.ts
T
DottaandPaperclip 5c7a025032 fix(ui): clear consumed messages from the pending queue (#16317)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Task conversations keep follow-up messages pending while an agent
works.
> - The screen combines optimistic sends, queue snapshots, and run input
records.
> - A late send response can restore a local queue flag after a run
takes the message.
> - A stale queue snapshot can then keep the same message in pending.
> - This pull request matches the input record to its source queue and
clears stale pending flags.
> - The benefit is that consumed messages leave pending while unsent
messages stay visible.

## Linked Issues or Issue Description

**What happened?**

A follow-up can remain in the task composer's pending row after a run
takes it as input. Local queue state can outlive the send request. The
old projection cleared that state for steering, but not for ordinary
run-start input. A stale queue read can also retain the consumed entry.

**Expected behavior**

A message that a started run takes as input appears once in history and
leaves pending. Other pending messages remain available. A message
assigned to a run that has not started remains pending.

**Steps to reproduce**

1. Send a follow-up while a task has a live run.
2. Let a started run record that message as input before or after the
send response arrives.
3. Let a queue read still include the consumed message.
4. Observe the message remain pending on the base revision.

**Paperclip version or commit**

Reproduced against master at `33d24e48ad` with controlled request
ordering.

**Deployment mode**

Source checkout. Browser verification used the production task screen
with a simulated native runner and delayed API responses. It did not run
a live model or change a deployed instance.

Related: [PR
#15030](https://github.com/paperclipai/paperclip/pull/15030) addresses
acknowledgement matching in the runner console. This change addresses
the task-detail composer and does not modify that console.

## What Changed

- Clear local pending flags when a started run records the comment as
input.
- Use live-run input IDs while the run-history query catches up.
- Match stale queue entries by their primary or coalesced source wake
ID, even when the snapshot targets the previous run or has no target.
- Retain coalesced source queue IDs in run context and include receipt
IDs in compact task run reads.
- Keep unstarted run inputs and authoritative retries pending. Cover
send ordering, same-run and successor delivery, queued snapshots,
coalesced inputs, and retry consumption.
- Add the delivery-race checks to the steering feature map.

## Verification

- Before the fix, the first four request-order regressions failed with
the consumed message still in pending. Coverage now includes 16
combinations of receipt source, send-response order, and stale snapshot
kind.
- Focused UI checks: 5 files and 233 tests passed. These cover task
detail, queue projection, optimistic comments, queued-message controls,
and the task-chat adapter.
- `pnpm -r typecheck` passed.
- Server receipt checks: 5 files and 139 tests passed with
`PAPERCLIP_TEST_POSTGRES_TEMPLATE=0`. These exercise real database
reads, coalesced heartbeat input, and polling routes. The default
cached-template path timed out during setup on this host; the
fresh-cluster path passed without longer timeouts.
- Token gates and `node scripts/check-maps.mjs` passed.
- Browser: sent a follow-up with a delayed response in the production
task screen. The consumed message moved into history. A second unsent
follow-up stayed queued. Reload preserved both states. The API and
runner were simulated.
- `pnpm build` passed.
- The full local repository test run was started, then stopped to
isolate database setup timeouts. All 55 GitHub CI checks passed on
`4ec6820cf1f89880b801fd5f938d72f54591c5e2`, including the full test
matrix. One serialized server shard passed on its single rerun after a
15-second route-test timeout; the same 97-test route suite also passed
locally.
- Greptile reviewed the final head at 5/5. All review threads are
resolved.

## Risks

- Queue display depends on matching comment IDs, source queue IDs, and a
recorded start time. Rows without start evidence stay pending. Newer
queues remain visible until their own input receipt arrives.
- Run context gains additive receipt metadata. No schema migration is
required. Dispatch order, stored queue state, company access, and
steering authority stay unchanged.
- Browser evidence covers controlled response ordering. It does not
establish deployment or live-provider behavior.

## Model Used

- OpenAI Codex, GPT-6, with reasoning, tool use, and code execution. The
exact serving model ID and context-window limit are not exposed in this
session.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` 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>
2026-10-14 16:11:29 -05:00

257 lines
8.8 KiB
TypeScript

import type { AgentAppearance } from "@paperclipai/shared";
import { HEARTBEAT_RUN_LIST_DEFAULT_LIMIT } from "@paperclipai/shared";
import type { IssueRecoveryAction } from "@paperclipai/shared";
import type {
HeartbeatRun,
HeartbeatRunEvent,
WorkspaceOperation,
ProviderTraceFrame,
ProviderTraceMetadata,
} from "@paperclipai/shared";
import { tenantSessionRecovery } from "@/lib/tenant-session-recovery";
import { api, type RequestOptions } from "./client";
export interface RunLivenessFields {
livenessState: HeartbeatRun["livenessState"];
livenessReason: string | null;
continuationAttempt: number;
lastUsefulActionAt: string | Date | null;
nextAction: string | null;
}
export interface ActiveRunForIssue {
execution?: HeartbeatRun["execution"];
id: string;
runtimeMode?: "legacy" | "native";
wakeupRequestId?: string | null;
queuedCommentSourceWakeIds?: string[] | null;
status: string;
invocationSource: string;
triggerDetail: string | null;
contextCommentId?: string | null;
contextWakeCommentId?: string | null;
startedAt: string | Date | null;
finishedAt: string | Date | null;
createdAt: string | Date;
agentId: string;
agentName: string;
agentAppearance?: AgentAppearance | null;
avatarUrl?: string;
adapterType: string;
logBytes?: number | null;
lastOutputBytes?: number | null;
issueId?: string | null;
livenessState?: RunLivenessFields["livenessState"];
livenessReason?: string | null;
continuationAttempt?: number;
lastUsefulActionAt?: string | Date | null;
nextAction?: string | null;
outputSilence?: HeartbeatRun["outputSilence"];
currentStatusMessage?: string | null;
currentStatusUpdatedAt?: string | Date | null;
currentToolName?: string | null;
lastAssistantSnippet?: string | null;
lastEventAt?: string | Date | null;
}
export interface LiveRunForIssue {
execution?: HeartbeatRun["execution"];
id: string;
runtimeMode?: "legacy" | "native";
wakeupRequestId?: string | null;
queuedCommentSourceWakeIds?: string[] | null;
status: string;
invocationSource: string;
triggerDetail: string | null;
contextCommentId?: string | null;
contextWakeCommentId?: string | null;
startedAt: string | null;
finishedAt: string | null;
createdAt: string;
agentId: string;
agentName: string;
agentAppearance?: AgentAppearance | null;
avatarUrl?: string;
adapterType: string;
logBytes?: number | null;
lastOutputBytes?: number | null;
issueId?: string | null;
livenessState?: RunLivenessFields["livenessState"];
livenessReason?: string | null;
continuationAttempt?: number;
lastUsefulActionAt?: string | null;
nextAction?: string | null;
outputSilence?: HeartbeatRun["outputSilence"];
currentStatusMessage?: string | null;
currentStatusUpdatedAt?: string | null;
currentToolName?: string | null;
lastAssistantSnippet?: string | null;
lastEventAt?: string | null;
}
export interface WatchdogDecisionInput {
runId: string;
decision: "snooze" | "continue" | "dismissed_false_positive";
evaluationIssueId?: string | null;
reason?: string | null;
snoozedUntil?: string | null;
}
export type RuntimeRequestKind =
| "runtime"
| "command_approval"
| "file_approval"
| "permission_approval"
| "user_input"
| "elicitation";
export type RuntimeRequestResolution =
| { action: "accept" | "accept_for_session" | "decline" | "cancel" }
| { action: "submit"; answers: Record<string, { answers: string[] }> }
| { action: "submit"; content: Record<string, unknown> }
| { action: "submit"; response: import("@paperclipai/adapter-utils").PaperclipQuestionResponse };
export interface HeartbeatRunListOptions {
summary?: boolean;
}
export interface ProviderTraceInspection {
trace: ProviderTraceMetadata | null;
entries: Array<Record<string, unknown>>;
}
export const heartbeatsApi = {
executionForIssue: (issueId: string) => api.get<{ runId: string; agentId: string; recoveryAction: IssueRecoveryAction | null; execution: HeartbeatRun["execution"] } | null>(`/issues/${issueId}/execution`),
list: (
companyId: string,
agentId?: string,
limit = HEARTBEAT_RUN_LIST_DEFAULT_LIMIT,
options: HeartbeatRunListOptions = {},
) => {
const searchParams = new URLSearchParams();
if (agentId) searchParams.set("agentId", agentId);
if (limit) searchParams.set("limit", String(limit));
if (options.summary) searchParams.set("summary", "true");
const qs = searchParams.toString();
return api.get<HeartbeatRun[]>(
`/companies/${companyId}/heartbeat-runs${qs ? `?${qs}` : ""}`,
);
},
get: (runId: string) => api.get<HeartbeatRun>(`/heartbeat-runs/${runId}`),
events: (runId: string, afterSeq = 0, limit = 200, options?: RequestOptions) =>
api.get<HeartbeatRunEvent[]>(
`/heartbeat-runs/${runId}/events?afterSeq=${encodeURIComponent(String(afterSeq))}&limit=${encodeURIComponent(String(limit))}`,
options,
),
log: (runId: string, offset = 0, limitBytes = 256000, options?: RequestOptions) =>
api.get<{
runId: string;
store: string;
logRef: string;
content: string;
nextOffset?: number;
}>(
`/heartbeat-runs/${runId}/log?offset=${encodeURIComponent(String(offset))}&limitBytes=${encodeURIComponent(String(limitBytes))}`,
options,
),
workspaceOperations: (runId: string) =>
api.get<WorkspaceOperation[]>(
`/heartbeat-runs/${runId}/workspace-operations`,
),
providerTrace: (runId: string) =>
api.get<ProviderTraceInspection>(`/heartbeat-runs/${runId}/provider-trace`),
providerTraceMetadata: (companyId: string, runIds: string[]) => {
const params = new URLSearchParams({ runIds: runIds.slice(0, 100).join(",") });
return api.get<ProviderTraceMetadata[]>(
`/companies/${companyId}/provider-traces?${params.toString()}`,
);
},
revealProviderTraceFrame: (runId: string, frameId: number) =>
api.post<ProviderTraceFrame>(
`/heartbeat-runs/${runId}/provider-trace/frames/${frameId}/reveal`,
{},
),
deleteProviderTrace: (runId: string) =>
api.delete<{ ok: true }>(`/heartbeat-runs/${runId}/provider-trace`),
downloadProviderTrace: async (runId: string): Promise<Blob> => {
const response = await fetch(
`/api/heartbeat-runs/${runId}/provider-trace/download`,
{
credentials: "include",
headers: { Accept: "application/x-ndjson" },
},
);
if (!response.ok) {
const body = (await response.json().catch(() => null)) as {
error?: string;
} | null;
const recovery = tenantSessionRecovery.recoverIfNeeded(response.status, body);
if (recovery) return recovery;
throw new Error(
body?.error ?? `Trace download failed: ${response.status}`,
);
}
return response.blob();
},
workspaceOperationLog: (
operationId: string,
offset = 0,
limitBytes = 256000,
) =>
api.get<{
operationId: string;
store: string;
logRef: string;
content: string;
nextOffset?: number;
}>(
`/workspace-operations/${operationId}/log?offset=${encodeURIComponent(String(offset))}&limitBytes=${encodeURIComponent(String(limitBytes))}`,
),
cancel: (runId: string) =>
api.post<void>(`/heartbeat-runs/${runId}/cancel`, {}),
resolveRuntimeRequest: (input: {
runId: string;
requestId: string;
turnId: string;
requestKind: RuntimeRequestKind;
resolution: RuntimeRequestResolution;
}) =>
api.post<{ accepted: true; commandId: string }>(
`/heartbeat-runs/${input.runId}/runtime-requests/${encodeURIComponent(input.requestId)}/resolve`,
{
turnId: input.turnId,
requestKind: input.requestKind,
resolution: input.resolution,
},
),
recordWatchdogDecision: (input: WatchdogDecisionInput) =>
api.post(`/heartbeat-runs/${input.runId}/watchdog-decisions`, {
decision: input.decision,
evaluationIssueId: input.evaluationIssueId ?? null,
reason: input.reason ?? null,
snoozedUntil: input.snoozedUntil ?? null,
}),
liveRunsForIssue: (issueId: string) =>
api.get<LiveRunForIssue[]>(`/issues/${issueId}/live-runs`),
activeRunForIssue: (issueId: string) =>
api.get<ActiveRunForIssue | null>(`/issues/${issueId}/active-run`),
liveRunsForCompany: (
companyId: string,
options?: number | { minCount?: number; limit?: number; distinctTasks?: boolean },
) => {
const searchParams = new URLSearchParams();
if (typeof options === "number") {
searchParams.set("minCount", String(options));
} else if (options) {
if (options.minCount)
searchParams.set("minCount", String(options.minCount));
if (options.limit) searchParams.set("limit", String(options.limit));
if (options.distinctTasks) searchParams.set("distinctTasks", "true");
}
const qs = searchParams.toString();
return api.get<LiveRunForIssue[]>(
`/companies/${companyId}/live-runs${qs ? `?${qs}` : ""}`,
);
},
};