diff --git a/docs/reference/slack-app-setup.md b/docs/reference/slack-app-setup.md index ed02c5a8e..de7d6cd42 100644 --- a/docs/reference/slack-app-setup.md +++ b/docs/reference/slack-app-setup.md @@ -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 diff --git a/packages/cli/src/commands/slack.ts b/packages/cli/src/commands/slack.ts index ce8e9ed84..d02dba294 100644 --- a/packages/cli/src/commands/slack.ts +++ b/packages/cli/src/commands/slack.ts @@ -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; incomplete?: Record }; 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}`); diff --git a/packages/daemon/src/domain/gateway/slack/message.ts b/packages/daemon/src/domain/gateway/slack/message.ts index efb9ce3d4..4a7f87524 100644 --- a/packages/daemon/src/domain/gateway/slack/message.ts +++ b/packages/daemon/src/domain/gateway/slack/message.ts @@ -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 diff --git a/packages/daemon/src/domain/gateway/slack/recovery-budget.ts b/packages/daemon/src/domain/gateway/slack/recovery-budget.ts index 73255c416..6471ef27b 100644 --- a/packages/daemon/src/domain/gateway/slack/recovery-budget.ts +++ b/packages/daemon/src/domain/gateway/slack/recovery-budget.ts @@ -19,6 +19,7 @@ export function budgetLeft(budget: RecoveryBudget, now: number): boolean { export class MethodBackoff { private readonly until = new Map(); 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)); } } diff --git a/packages/daemon/src/domain/gateway/slack/settled-asks.ts b/packages/daemon/src/domain/gateway/slack/settled-asks.ts new file mode 100644 index 000000000..6c98d70ea --- /dev/null +++ b/packages/daemon/src/domain/gateway/slack/settled-asks.ts @@ -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(); + private readonly backoff = new MethodBackoff(); + private budget = newRecoveryBudget(Date.now()); + private readonly controller = new AbortController(); + private unsubscribe?: () => void; + private timer?: ReturnType; + 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 { + 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): Promise { + 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 { + 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; + } +} diff --git a/packages/daemon/src/domain/gateway/slack/slack-subsystem.ts b/packages/daemon/src/domain/gateway/slack/slack-subsystem.ts index a51f08b9c..2d819ba67 100644 --- a/packages/daemon/src/domain/gateway/slack/slack-subsystem.ts +++ b/packages/daemon/src/domain/gateway/slack/slack-subsystem.ts @@ -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 }, diff --git a/packages/daemon/src/domain/queue-repository.ts b/packages/daemon/src/domain/queue-repository.ts index 9dd67e56d..bb905f116 100644 --- a/packages/daemon/src/domain/queue-repository.ts +++ b/packages/daemon/src/domain/queue-repository.ts @@ -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) => { diff --git a/packages/daemon/src/domain/queue-transition-log.ts b/packages/daemon/src/domain/queue-transition-log.ts index 0edb7765d..a1a254511 100644 --- a/packages/daemon/src/domain/queue-transition-log.ts +++ b/packages/daemon/src/domain/queue-transition-log.ts @@ -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; } diff --git a/packages/daemon/test/slack-settled-ask.test.ts b/packages/daemon/test/slack-settled-ask.test.ts index e10cfbb7e..3c4a57c8c 100644 --- a/packages/daemon/test/slack-settled-ask.test.ts +++ b/packages/daemon/test/slack-settled-ask.test.ts @@ -29,6 +29,8 @@ describe("posted asks follow settlement outside Slack", () => { let messages: Message[]; let updates: Array>; let postCount: number; + let apiHook: ((method: string, body: Record) => 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 : {}; + 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); + }); });