fix(slack): settle original asks when queue work moves on

This commit is contained in:
dev-guard committed 2026-11-04 04:53:15 +00:00
1 parent 21d4e51a6e
commit dfdb65a7c9
9 files changed
+370 -9

No files matched your search

+20
View File
@@ -366,6 +366,26 @@ Quiet hours hold a person's routine asks while they're on, and post them once th
## Reconnect recovery and status
When a posted ask is answered, closed, canceled or replaced elsewhere, the connector edits its
original primary message with the status, time, actor and closure note. Buttons become option
text and the answer hint disappears; the remaining content and reconciliation marker stay as
posted. Supplemental messages are unchanged. This adds no post or reaction and needs no agent
turn. Separate resolution notices retain their existing behavior.
Edits follow committed queue changes, with a startup catch-up for posted asks on retained queue
rows. Each primary costs one exact-message read and one `chat.update`. The serial edit worker
uses the recovery budget helpers (four calls, 100 entries, 15 seconds) and waits for the next
budget window when exhausted. Slack's per-method `Retry-After` is retained across restarts.
Only an acknowledged edit gets a success receipt. `rig slack status` shows queued reconciliation
and the last edit failure; a refused read, failed write or lost response stays unfinished and
is retried. These edit calls have their own budget, separate from missed-message scanning below.
The baseline already includes public-channel history access and `chat:write`. Private-channel
reads may require `groups:history`, which the baseline does not grant; a refused read never
counts as an edit. An unavailable/deleted original cannot be replaced by a new post. Startup
catch-up cannot edit rows already removed by queue retention or posts without their original
notification receipt and channel stamp.
The daemon scans available top-level messages and replies broadcast with **Also send to
channel** in the channels this machine owns: its default channel and its channel map's, except a
channel it relays to another machine. It also scans replies in the threads it tracks in those
+5
View File
@@ -251,6 +251,7 @@ export function slackCommand(deps: SlackDeps = {}): Command {
log(`Daemon: ${observation.state}${observation.reason ? ` (${observation.reason})` : ""}`);
const connector = observation.connector as { configurationDigest?: string; inboundReady?: boolean; outboundReady?: boolean;
inboundRouting?: { fallback: string; destination: string };
settlement?: { pending: number; lastFailure: string | null };
deadLetterBacklog?: number | null; deadLetterBacklogState?: string; deadLetterBacklogReason?: string;
inbound?: InboundDeliveryStatus & { state?: string; generation?: number; lastEventAt?: string };
recovery?: { state?: string; reason?: string; lastScanAt?: string; acceptedThisProcess?: number; deadLetteredThisProcess?: number;
@@ -261,6 +262,10 @@ export function slackCommand(deps: SlackDeps = {}): Command {
backoff?: Record<string, number>; incomplete?: Record<string, number> };
limits?: string[] } } | undefined;
if (connector) {
if (connector.settlement) {
log(` Ask edits: ${connector.settlement.pending} queued for reconciliation`);
if (connector.settlement.lastFailure) log(` Last ask-edit failure: ${connector.settlement.lastFailure}`);
}
if (connector.inboundRouting?.fallback === "local") log(` Configured routing fallback: local to ${connector.inboundRouting.destination} (after forwarding, thread and channel routes)`);
log(` Socket: ${connector.inbound?.state ?? "unknown"}; generation ${connector.inbound?.generation ?? "unknown"}; last event ${connector.inbound?.lastEventAt ?? "unknown"}`);
for (const line of socketDeliveryLines(connector.inbound)) log(` ${line}`);
@@ -134,8 +134,9 @@ export function buildImageBlocks(mediaRefs: readonly SlackMediaRef[] | undefined
* exactly these (one producer, one parser: see parseQuestionAction). */
export const QUESTION_BLOCK_PREFIX = "or-q:";
export const OPTION_ACTION_PREFIX = "or-opt:";
const TYPED_REPLY_HINT = "Or reply in this thread with your own answer. A typed reply submits your answer without waiting for every question to be answered.";
const CHANNEL_REPLY_HINT = "Reply in this thread with your answer.";
export const TYPED_REPLY_HINT = "Or reply in this thread with your own answer. A typed reply submits your answer without waiting for every question to be answered.";
export const CHANNEL_REPLY_HINT = "Reply in this thread with your answer.";
export const HUMAN_REQUIRED_REPLY_HINT = "Reply in this thread after this question to answer it. Earlier and extra messages still reach the agent as conversation.";
/** Parse a clicked button back into its question and option ids; null if it is not ours. */
export function parseQuestionAction(blockId: unknown, actionId: unknown): { questionId: string; optionId: string } | null {
@@ -597,7 +598,7 @@ export function buildOutboundMessage(q: QitemLike, opts: OutboundMessageOpts): S
throw new HumanMessageShapeError("Extra blocks have no complete accessible fallback. Use mediaRefs for images or author supplemental human detail.");
}
const replyHint = q.ownerNotificationKind === "human-required"
? "Reply in this thread after this question to answer it. Earlier and extra messages still reach the agent as conversation." : null;
? HUMAN_REQUIRED_REPLY_HINT : null;
const fallback = (context: string | null, repeat: string | null = null) => [attr, headline, repeat, context, body, questionParts?.text, replyHint, evidence ? evidence.text : null, attachmentText, opts.reconcileMarker].filter(Boolean).join("\n");
// The gate context line is supplementary: when it alone would push the complete fallback over
// budget, the instance part alone is tried, and the line is left out only when even that
@@ -19,6 +19,7 @@ export function budgetLeft(budget: RecoveryBudget, now: number): boolean {
export class MethodBackoff {
private readonly until = new Map<string, number>();
blocked(method: string, now: number): boolean { return (this.until.get(method) ?? 0) > now; }
waitMs(method: string, now: number): number { return Math.max(0, (this.until.get(method) ?? 0) - now); }
hold(method: string, at: number): void { this.until.set(method, Math.max(at, this.until.get(method) ?? 0)); }
}
@@ -0,0 +1,196 @@
// Edit the posted primary, never re-render an old ask from its mutable queue body.
// Queue receipts are the durable backlog; successful edits alone get an edited receipt.
import type { QueueRepository } from "../../queue-repository.js";
import type { QueueTransition } from "../../queue-transition-log.js";
import { callWebApi, type FetchImpl, type WebApiResult } from "./slack-api.js";
import { budgetLeft, HISTORY, REPLIES, MethodBackoff, newRecoveryBudget } from "./recovery-budget.js";
import { parsePostedStamp } from "./thread-seat-map.js";
import { TYPED_REPLY_HINT, CHANNEL_REPLY_HINT, HUMAN_REQUIRED_REPLY_HINT, QUESTION_BLOCK_PREFIX, escapeSlackText, redactSecrets } from "./message.js";
const UPDATE = "chat.update";
const EDITED = "slack-ask-settled";
const RATE_LIMIT = "slack-ask-edit-rate-limit";
const STATUS_BLOCK = "or-settled";
const hints = new Set([TYPED_REPLY_HINT, CHANNEL_REPLY_HINT, HUMAN_REQUIRED_REPLY_HINT]);
const terminal = new Set(["done", "canceled", "denied", "failed", "handed-off"]);
const field = (note: string, key: string) => note.split(/\s+/).find((part) => part.startsWith(`${key}=`))?.slice(key.length + 1);
type Block = { type?: string; block_id?: string; elements?: Array<{ text?: string }>; [key: string]: unknown };
/** Remove only our answer controls, retaining authored content and reconciliation identity. */
function handledMessage(message: { text: string; blocks: Block[] }, status: string) {
let text = message.text;
const previousStatus = message.blocks.find((b) => b.block_id === STATUS_BLOCK)?.elements?.[0]?.text;
if (previousStatus && text.startsWith(previousStatus + "\n")) text = text.slice(previousStatus.length + 1);
const blocks = message.blocks.flatMap((block): Block[] => {
if (block.block_id === STATUS_BLOCK) return [];
if (block.type === "actions" && block.block_id?.startsWith(QUESTION_BLOCK_PREFIX)) {
// The labels remain history after their controls disappear, as in the resolved-notice render.
const buttons = block.elements as unknown as Array<{ text?: { text?: string }; style?: string }>;
return [{ type: "section", text: { type: "mrkdwn", text: buttons.map((b) => `• ${b.text?.text ?? ""}${b.style === "primary" ? " (recommended)" : ""}`).join("\n") } }];
}
const hint = block.type === "context" && block.elements?.length === 1 ? block.elements[0]?.text : undefined;
if (!hint || !hints.has(hint)) return [block];
// This generated hint's last occurrence, never a matching sentence earlier in the brief.
const at = text.lastIndexOf("\n" + hint);
if (at >= 0) text = text.slice(0, at) + text.slice(at + hint.length + 1);
return [];
});
return { text: status + "\n" + text,
blocks: [{ type: "context", block_id: STATUS_BLOCK, elements: [{ type: "mrkdwn", text: status }] }, ...blocks] };
}
function ending(transitions: QueueTransition[], episode: QueueTransition): QueueTransition | undefined {
return transitions.find((t) => t.transitionId > episode.transitionId && (
terminal.has(t.state) || t.ownerNotificationKind === "human-decision-resolved"
|| t.ownerNotificationKind === "human-required"
|| (episode.state === "blocked" && t.state !== "blocked")
|| (t.transitionNote ?? "").startsWith("fallback-routed:")
));
}
function statusLine(t: QueueTransition): string {
const what = t.ownerNotificationKind === "human-decision-resolved" ? "Answered" : t.state === "done" ? "Done"
: t.state === "canceled" ? "Canceled" : terminal.has(t.state) ? t.state : "Handled";
const note = redactSecrets((t.transitionNote ?? "").replace(/\s+/g, " ")).slice(0, 1000);
return escapeSlackText([`${what}${t.closureReason ? ` · ${t.closureReason}` : ""}`, t.ts, `by ${t.actorSession}`, note].filter(Boolean).join(" · "));
}
export class SlackAskSettlement {
private readonly pending = new Map<string, number>();
private readonly backoff = new MethodBackoff();
private budget = newRecoveryBudget(Date.now());
private readonly controller = new AbortController();
private unsubscribe?: () => void;
private timer?: ReturnType<typeof setTimeout>;
private running = false;
private stopped = true;
private lastFailure: string | null = null;
constructor(private readonly opts: { queue: QueueRepository; token: string; fetchImpl?: FetchImpl; log: (message: string) => void }) {}
start(): void {
this.stopped = false;
this.unsubscribe = this.opts.queue.subscribeChanges((id) => this.enqueue(id));
for (const id of this.opts.queue.transitionLog.postedHumanAskIds()) {
for (const transition of this.opts.queue.transitionLog.listForQitem(id)) {
const note = transition.transitionNote ?? "";
if (!note.startsWith(RATE_LIMIT + " ")) continue;
const method = field(note, "method");
const until = Number(field(note, "retry_at"));
if (method && Number.isFinite(until)) this.backoff.hold(method, until);
}
this.enqueue(id);
}
}
stop(): void {
this.stopped = true;
this.unsubscribe?.();
if (this.timer) clearTimeout(this.timer);
this.controller.abort();
}
status() { return { pending: this.pending.size, lastFailure: this.lastFailure }; }
private enqueue(id: string): void {
if (this.stopped) return;
if (!this.pending.has(id)) this.pending.set(id, 0);
if (!this.running && !this.timer) this.schedule(0);
}
private schedule(ms: number): void {
if (this.stopped) return;
this.timer = setTimeout(() => { this.timer = undefined; void this.pump(); }, Math.max(0, ms));
this.timer.unref?.();
}
private async pump(): Promise<void> {
this.running = true;
try {
while (!this.stopped && this.pending.size) {
const now = Date.now();
if (now >= this.budget.deadline) this.budget = newRecoveryBudget(now);
if (!budgetLeft(this.budget, now)) { this.schedule(this.budget.deadline - now); return; }
const next = [...this.pending].find(([, at]) => at <= now);
if (!next) { this.schedule(Math.min(...this.pending.values()) - now); return; }
const id = next[0];
const wait = await this.settleOne(id);
if (wait === null) this.pending.delete(id);
else this.pending.set(id, Date.now() + wait);
}
} catch (error) {
this.lastFailure = `settlement retained: ${(error as Error).message}`;
this.opts.log(this.lastFailure);
this.schedule(30_000);
} finally { this.running = false; }
}
private async api(method: string, body: Record<string, unknown>): Promise<WebApiResult> {
this.budget.calls++;
const fetchImpl: FetchImpl = (url, init) => (this.opts.fetchImpl ?? fetch)(url, { ...init,
signal: init?.signal ? AbortSignal.any([init.signal, this.controller.signal]) : this.controller.signal });
return callWebApi(method, this.opts.token, body, fetchImpl, Math.max(1, this.budget.deadline - Date.now()));
}
private failed(id: string, method: string, result: WebApiResult): number {
const wait = Math.max(30_000, (result.retryAfterSeconds ?? 0) * 1000);
if (result.status === 429) {
const until = Date.now() + wait;
this.backoff.hold(method, until);
this.opts.queue.update({ qitemId: id, actorSession: "daemon@kernel",
transitionNote: `${RATE_LIMIT} method=${method} retry_at=${until}` });
}
this.lastFailure = `${id}: ${method} not done (${result.error ?? result.status})`;
this.opts.log(`slack ask settlement retained: ${this.lastFailure}`);
return wait;
}
/** null = no remaining edits; 0 = next primary; positive = retained retry. */
private async settleOne(id: string): Promise<number | null> {
if (!this.opts.queue.db.prepare("SELECT 1 FROM queue_items WHERE qitem_id = ?").get(id)) return null;
const transitions = this.opts.queue.transitionLog.listForQitem(id);
for (const receipt of transitions) {
const note = receipt.transitionNote ?? "";
if (!note.startsWith("slack-owner-notification-posted ") || field(note, "kind") !== "human-required" || field(note, "digest")) continue;
const key = field(note, "notification_key");
const episode = transitions.find((t) => `${id}:${t.transitionId}` === key);
if (!episode) continue;
const end = ending(transitions, episode);
if (!end) continue;
const messageTs = field(note, "message_ts");
const stamp = transitions.filter((t) => t.transitionId < receipt.transitionId)
.map((t) => parsePostedStamp(t.transitionNote ?? ""))
.filter((s) => s?.messageTs === messageTs && s.origin === "posted").at(-1);
if (!stamp) continue; // No authoritative channel/timestamp: never guess a replacement post.
const edited = `${EDITED} notification_key=${key} channel=${stamp.channel} message_ts=${messageTs}`;
if (transitions.some((t) => t.transitionNote === edited)) continue;
if (this.budget.calls + 2 > 4) return Math.max(1, this.budget.deadline - Date.now());
const method = stamp.messageTs === stamp.threadTs ? HISTORY : REPLIES;
const held = Math.max(this.backoff.waitMs(method, Date.now()), this.backoff.waitMs(UPDATE, Date.now()));
if (held) return held;
// Exact original message, including on an upgrade with only the old posted stamps.
const read = await this.api(method, { channel: stamp.channel, limit: 1, inclusive: true,
...(method === HISTORY ? { latest: messageTs } : { ts: stamp.threadTs, oldest: messageTs }) });
if (this.stopped) return null;
if (!read.ok) return this.failed(id, method, read);
this.budget.entries++;
const message = (Array.isArray(read.json.messages) ? read.json.messages : []).find((m) => m?.ts === messageTs);
if (!message || typeof message.text !== "string" || !Array.isArray(message.blocks)) {
return this.failed(id, method, { ...read, error: "original message unavailable" });
}
if (!budgetLeft(this.budget, Date.now())) return 0;
// A queue change during the read may add information; derive again before writing.
const currentEnd = ending(this.opts.queue.transitionLog.listForQitem(id), episode);
if (!currentEnd || this.stopped) return null;
const update = await this.api(UPDATE, { channel: stamp.channel, ts: messageTs, ...handledMessage(message, statusLine(currentEnd)) });
if (this.stopped) return null;
if (!update.ok) return this.failed(id, UPDATE, update);
if (update.json.ts !== messageTs || update.json.channel !== stamp.channel) {
return this.failed(id, UPDATE, { ...update, error: "update acknowledgement did not name the requested message" });
}
this.opts.queue.update({ qitemId: id, actorSession: "daemon@kernel", transitionNote: edited });
return 0;
}
return null;
}
}
@@ -26,6 +26,7 @@ import { resolveSecret } from "./secrets.js";
import { SeenStore, DeadLetterStore, InboundReceiptStore, slackMicros } from "./state-store.js";
import { makeQueuePorts } from "./queue-access.js";
import { SlackOutboundDriver, OUTBOUND_OP, type OutboundPostPayload } from "./outbound-driver.js";
import { SlackAskSettlement } from "./settled-asks.js";
import { abortableSleep, subsystemSlackDeliver } from "./slack-delivery.js";
import { InboundRouter, SLACK_THREAD_ROOT_TAG, handleEnvelope, type SocketEnvelope, type SlackEvent, type SlackBlockActions, type InboundFilePort, type InboundFileResult, type StoredInboundFile, type FailedInboundFile } from "./inbound.js";
import { makeInboundSenderResolver, type RegistrySurface } from "./inbound-admission.js";
@@ -1006,6 +1007,11 @@ export function buildSlackGatewayWire(opts: SlackWireOpts): GatewayWire {
// hermeticity rule, same as every monitor in the supervision tree.
const stops: Array<() => void> = [];
const starts: Array<() => void> = [];
const settlement = outboundReady ? new SlackAskSettlement({ queue: opts.queueRepo, token: bot!, fetchImpl: opts.fetchImpl, log }) : null;
if (settlement) {
starts.push(() => settlement.start());
stops.push(() => settlement.stop());
}
let inboundHandle: SocketInboundHandle | undefined;
let recovery: SlackRecovery | undefined;
// Recovery covers the channels this machine owns and the threads tracked in them. The configured
@@ -1200,6 +1206,7 @@ export function buildSlackGatewayWire(opts: SlackWireOpts): GatewayWire {
platform: "slack", configurationDigest: channelStateDigest(cfg),
outboundReady,
inboundReady,
...(settlement ? { settlement: settlement.status() } : {}),
inboundRouting,
recovery: recovery?.status() ?? { state: "unavailable", reason: "inbound-not-configured" },
inbound: inboundHandle?.status() ?? { state: inboundReady ? "not-started" : "not-configured", generation: 0, reconnects: 0 },
@@ -1081,6 +1081,13 @@ export class QueueRepository {
refreshQueueWaits(this.db, this.watchdogJobsRepo ?? new WatchdogJobsRepository(this.db), changedQitem, proofChanged);
}
/** Observe committed queue changes; callbacks should schedule work, not perform it inline. */
subscribeChanges(onChange: (qitemId: string) => void): () => void {
return this.eventBus.subscribe((event) => {
if (event.type.startsWith("queue.") && "qitemId" in event && typeof event.qitemId === "string") onChange(event.qitemId);
});
}
startWaitReminders(): () => void {
this.reconcileWaitReminders();
return this.eventBus.subscribe((event) => {
@@ -454,6 +454,15 @@ export class QueueTransitionLog {
return byQitem;
}
/** Startup catch-up includes retained terminal rows, unlike active-episode selection. */
postedHumanAskIds(): string[] {
if (!this.hasOwnerNotificationColumns) return [];
return (this.db.prepare(`SELECT DISTINCT qitem_id FROM ${this.historySource}
WHERE transition_note LIKE 'slack-owner-notification-posted %'
AND qitem_id IN (SELECT qitem_id FROM queue_items)
AND instr(' ' || transition_note || ' ', ' kind=human-required ') > 0`).all() as Array<{ qitem_id: string }>).map((r) => r.qitem_id);
}
hasOwnerNotificationReceipt(qitemId: string, notificationKey: string): boolean {
return this.getOwnerNotificationReceipt(qitemId, notificationKey) !== null;
}
+121 -6
View File
@@ -29,6 +29,8 @@ describe("posted asks follow settlement outside Slack", () => {
let messages: Message[];
let updates: Array<Record<string, unknown>>;
let postCount: number;
let apiHook: ((method: string, body: Record<string, unknown>) => Response | undefined) | undefined;
let methods: string[];
const logs: string[] = [];
function start() {
@@ -37,6 +39,10 @@ describe("posted asks follow settlement outside Slack", () => {
fetchImpl: async (url, init) => {
const api = new URL(url);
const body = init?.body ? JSON.parse(String(init.body)) as Record<string, unknown> : {};
const method = api.pathname.split("/").at(-1)!;
methods.push(method);
const override = apiHook?.(method, body);
if (override) return override;
if (api.pathname.endsWith("chat.postMessage")) {
postCount++;
const message = { ...body, ts: `${postCount}.000001` } as Message;
@@ -66,29 +72,32 @@ describe("posted asks follow settlement outside Slack", () => {
const secrets = join(home, "fake.env");
writeFileSync(secrets, "SLACK_BOT_TOKEN=xoxb-EXAMPLE-fake\n");
saveConfig({ ...DEFAULT_CONFIG, enabled: true, channel: "C-ASKS", secretsEnvFile: secrets }, home);
messages = []; updates = []; postCount = 0; logs.length = 0;
messages = []; updates = []; postCount = 0; logs.length = 0; methods = []; apiHook = undefined;
start();
});
afterEach(() => { wire.stop(); db.close(); rmSync(home, { recursive: true, force: true }); });
afterEach(() => { wire.stop(); vi.useRealTimers(); db.close(); rmSync(home, { recursive: true, force: true }); });
async function postAsk(detail?: string) {
const before = postCount;
const item = await repo.create({ sourceSession: "author@rig", destinationSession: human,
summary: "Ship the change?", body: "Original **brief** with its evidence.", humanDetail: detail,
humanQuestions: [{ id: "ship", question: "Ship now?", options: [{ id: "yes", label: "Ship" }, { id: "hold", label: "Hold" }] }], nudge: false });
const [alert] = await makeQueuePorts(repo, { loadHumanRegistry: () => registry }).listHumanAlerts({});
expect(alert?.qitemId).toBe(item.qitemId);
expect(wire.dispatcher.dispatch("post_message", human, alert)).toMatchObject({ ok: true });
await vi.waitFor(() => expect(repo.currentHumanRequiredEpisode(item.qitemId)?.receipt?.messageTs).toBe("1.000001"));
expect(messages[0]!.blocks.some((b) => b.type === "actions")).toBe(true);
expect(messages[0]!.text).toContain("Reply in this thread after this question");
await vi.waitFor(() => expect(repo.currentHumanRequiredEpisode(item.qitemId)?.receipt?.messageTs).toBe(`${before + 1}.000001`));
expect(messages[before]!.blocks.some((b) => b.type === "actions")).toBe(true);
expect(messages[before]!.text).toContain("Reply in this thread after this question");
return item;
}
it.each(["done", "canceled"] as const)("edits the original primary after a CLI-style %s transition, preserving its content and marker", async (state) => {
const item = await postAsk();
const original = structuredClone(messages[0]!);
const successor = state === "canceled" ? await repo.create({ sourceSession: "author@rig", destinationSession: "worker@rig",
summary: "Replacement work", body: "The successor owns the remaining work.", nudge: false }) : undefined;
repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state,
closureReason: "superseded", transitionNote: "Handled in the pane after the merge." });
closureReason: "superseded", ...(successor ? { closureTarget: successor.qitemId } : {}), transitionNote: "Handled in the pane after the merge." });
const closed = repo.listTransitions(item.qitemId).at(-1)!;
expect(repo.getById(item.qitemId)?.state).toBe(state);
await vi.waitFor(() => expect(updates).toHaveLength(1));
@@ -143,4 +152,110 @@ describe("posted asks follow settlement outside Slack", () => {
expect(messages[0]!.blocks.some((b) => b.type === "actions")).toBe(true);
expect(postCount).toBe(1);
});
it("edits the original on a typed answer and names the answering person", async () => {
const item = await postAsk();
expect(repo.resolveDirectHumanReply({ qitemId: item.qitemId, actorSession: human, decision: "Ship it" })).toBe(true);
await vi.waitFor(() => expect(updates).toHaveLength(1));
expect(updates[0]!.text).toContain("Answered");
expect(updates[0]!.text).toContain(`by ${human}`);
expect(postCount).toBe(1);
});
it("retains a refused private-channel read without claiming the edit, then recovers on restart", async () => {
const item = await postAsk();
apiHook = (method) => method === "conversations.history" ? response({ ok: false, error: "missing_scope" }) : undefined;
repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" });
await vi.waitFor(() => expect(logs.join("\n")).toContain("missing_scope"));
expect(methods).not.toContain("chat.update");
expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settled "))).toBe(false);
expect(wire.status?.()).toMatchObject({ settlement: { pending: 1, lastFailure: expect.stringContaining("not done") } });
wire.stop(); apiHook = undefined; start();
await vi.waitFor(() => expect(updates).toHaveLength(1));
expect(postCount).toBe(1);
});
it("honors a chat.update Retry-After, without a success receipt until the retry succeeds", async () => {
const item = await postAsk();
wire.stop(); vi.useFakeTimers();
let limited = true;
apiHook = (method) => method === "chat.update" && limited
? new Response(JSON.stringify({ ok: false, error: "ratelimited" }), { status: 429, headers: { "retry-after": "60" } }) : undefined;
repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" });
start();
await vi.advanceTimersByTimeAsync(0);
expect(methods.filter((m) => m === "chat.update")).toHaveLength(1);
expect(updates).toHaveLength(0);
expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settled "))).toBe(false);
wire.stop(); start(); // Retry-After survives the wire, like the posted receipt.
await vi.advanceTimersByTimeAsync(0);
expect(methods.filter((m) => m === "chat.update")).toHaveLength(1);
await vi.advanceTimersByTimeAsync(59_999);
expect(methods.filter((m) => m === "chat.update")).toHaveLength(1);
limited = false;
await vi.advanceTimersByTimeAsync(1);
expect(updates).toHaveLength(1);
expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settled "))).toBe(true);
expect(postCount).toBe(1);
});
it("replays an edit whose HTTP result was lost without duplicating status or posting anew", async () => {
const item = await postAsk();
apiHook = (method, body) => {
if (method !== "chat.update") return;
Object.assign(messages[0]!, body); // Slack applied it, but its answer was lost.
throw new Error("response lost");
};
repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" });
await vi.waitFor(() => expect(logs.join("\n")).toContain("response lost"));
expect(repo.listTransitions(item.qitemId).some((t) => t.transitionNote?.startsWith("slack-ask-settled "))).toBe(false);
wire.stop(); apiHook = undefined; start();
await vi.waitFor(() => expect(updates).toHaveLength(1));
expect(messages[0]!.blocks.filter((b) => b.block_id === "or-settled")).toHaveLength(1);
expect(messages[0]!.text.match(/^Done/gm)).toHaveLength(1);
expect(postCount).toBe(1);
});
it("bounds a closure burst to the recovery call budget and drains the remaining primary later", async () => {
const items = [await postAsk(), await postAsk(), await postAsk()];
wire.stop(); vi.useFakeTimers();
for (const item of items) repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" });
start();
await vi.advanceTimersByTimeAsync(0);
expect(updates).toHaveLength(2);
expect(methods.filter((m) => m === "conversations.history")).toHaveLength(2);
await vi.advanceTimersByTimeAsync(14_999);
expect(updates).toHaveLength(2);
await vi.advanceTimersByTimeAsync(1);
expect(updates).toHaveLength(3);
expect(new Set(updates.map((m) => m.ts)).size).toBe(3);
expect(postCount).toBe(3);
});
it("unparks an answered gate and keeps its next thread-primary interactive until that park ends", async () => {
const item = await repo.create({ sourceSession: "author@rig", destinationSession: "worker@rig",
summary: "Choose the next step", body: "Original work", nudge: false });
const park = () => repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "blocked", blockedOn: human,
summary: "May I continue?", evidenceRef: "https://example.com/proof", transitionNote: "Need the person's answer." });
const send = async (ts: string) => {
const [alert] = await makeQueuePorts(repo, { loadHumanRegistry: () => registry }).listHumanAlerts({});
expect(alert?.qitemId).toBe(item.qitemId);
expect(wire.dispatcher.dispatch("post_message", human, alert)).toMatchObject({ ok: true });
await vi.waitFor(() => expect(repo.currentHumanRequiredEpisode(item.qitemId)?.receipt?.messageTs).toBe(ts));
};
park(); await send("1.000001");
expect(messages[0]!.text).toContain("Reply in this thread after this question");
repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "in-progress", transitionNote: "Answered in the pane; continuing." });
await vi.waitFor(() => expect(updates).toHaveLength(1));
park(); await send("2.000001");
expect(messages[1]!.thread_ts).toBe("1.000001");
expect(messages[1]!.text).toContain("Reply in this thread after this question");
expect(updates).toHaveLength(1);
repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "in-progress", transitionNote: "Second answer received." });
await vi.waitFor(() => expect(updates).toHaveLength(2));
expect(updates[1]!.ts).toBe("2.000001");
expect(methods).toContain("conversations.replies");
expect(messages[1]!.text).not.toContain("Reply in this thread after this question");
expect(postCount).toBe(2);
});
});