diff --git a/docs/reference/slack-app-setup.md b/docs/reference/slack-app-setup.md index ed02c5a8e..a64831c39 100644 --- a/docs/reference/slack-app-setup.md +++ b/docs/reference/slack-app-setup.md @@ -366,6 +366,36 @@ 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. + +A deliberately reopened direct ask says **Open again**; its options remain as text. If it has +no committed answer, it tells the person to reply in its thread. Otherwise it says **Earlier +answer retained**: reopening does not accept a second answer. Earlier replaced parks remain +historical; **Asked again** does not claim that the later ask has been posted. Edits reconcile +the current state again after awaited Slack calls, so an obsolete close is not recorded as the +current result after a reopen. + +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. Transient failures retry with increasing waits, capped at an hour, +behind other queued work. Permanent Slack failures (such as missing scope, a deleted original, +an archived channel or an uneditable message) remain unfinished and are retried only on the next +connector start. They do not keep consuming edit capacity. 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..6acdf2c3d --- /dev/null +++ b/packages/daemon/src/domain/gateway/slack/settled-asks.ts @@ -0,0 +1,243 @@ +// 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 { createHash } from "node:crypto"; +import { resolveRegisteredHumanAddress } from "../human-registry.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 permanentErrors = new Set(["missing_scope", "not_in_channel", "channel_not_found", "is_archived", + "cant_update_message", "invalid_blocks", "message_not_found", "thread_not_found", "original message unavailable"]); +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 statusLine(t: QueueTransition, label?: string): string { + const what = label ?? (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 failures = new Map(); + private readonly unavailable = new Set(); + 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.delete(id); 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()), + method === HISTORY || method === REPLIES ? "get-query" : "json-post"); + } + + private failed(id: string, key: string, method: string, result: WebApiResult): number { + this.lastFailure = `${id}: ${method} not done (${result.error ?? result.status})`; + this.opts.log(`slack ask settlement retained: ${this.lastFailure}`); + if (permanentErrors.has(result.error ?? "")) { this.unavailable.add(key); return 0; } + const count = (this.failures.get(key) ?? 0) + 1; + this.failures.set(key, count); + const wait = Math.max(Math.min(30_000 * 2 ** Math.min(count - 1, 7), 3_600_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}` }); + } + return wait; + } + + private desired(id: string, episode: QueueTransition, recipient: string, transitions: QueueTransition[]) { + const item = this.opts.queue.getById(id); + if (!item) return undefined; + const after = transitions.filter((t) => t.transitionId > episode.transitionId && ![ + "slack-posted ", "slack-owner-notification-posted ", EDITED + " ", RATE_LIMIT + " ", + ].some((prefix) => (t.transitionNote ?? "").startsWith(prefix))); + const make = (t: QueueTransition, label?: string) => { + const status = statusLine(t, label); + return { status, revision: createHash("sha256").update(status).digest("hex").slice(0, 24) }; + }; + const exit = after.find((t) => terminal.has(t.state) || t.ownerNotificationKind === "human-decision-resolved" + || (episode.state === "blocked" && t.state !== "blocked")); + const replacement = after.find((t) => t.ownerNotificationKind === "human-required"); + if (replacement) return exit && exit.transitionId < replacement.transitionId + ? make(exit) : make(replacement, "Asked again"); + const rerouted = after.find((t) => (t.transitionNote ?? "").startsWith("fallback-routed:")); + if (rerouted) return make(rerouted, "Handled"); + if (terminal.has(item.state)) { + const latest = after.filter((t) => t.state === item.state).at(-1); + return latest ? make(latest) : undefined; + } + const registry = this.opts.queue.humanRegistry(); + const samePerson = (address: string | null) => address === recipient || (registry.ok && + resolveRegisteredHumanAddress(recipient, registry.entities) !== null && + resolveRegisteredHumanAddress(address, registry.entities) === resolveRegisteredHumanAddress(recipient, registry.entities)); + if (episode.state === "pending" && samePerson(item.destinationSession)) { + const reopened = after.filter((t) => (t.transitionNote ?? "").startsWith("reopen acknowledged:")).at(-1); + if (reopened) { + const answered = this.opts.queue.getHumanAnswer(id, recipient) ?? this.opts.queue.committedAnswerFromAnother(id, recipient); + return make(reopened, answered ? "Open again · Earlier answer retained" : "Open again · Reply in this thread to answer"); + } + } + // A reopened work row does not reopen the older park it already left. A new human park has its own episode above. + if (exit) return make(exit); + if (episode.state === "blocked" && item.state === "blocked" && registry.ok && !samePerson(item.blockedOn)) { + const moved = after.at(-1); + if (moved) return make(moved, "Handled"); + } + return undefined; + } + + /** 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 messageTs = field(note, "message_ts"); + const stamp = transitions.filter((t) => t.transitionId < receipt.transitionId) + .map((t) => parsePostedStamp(t.transitionNote ?? "")) + .filter((s) => s !== null && 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 (this.unavailable.has(edited)) continue; + const desired = this.desired(id, episode, stamp.human, transitions); + if (!desired) continue; + const lastAck = transitions.filter((t) => t.transitionNote === edited || t.transitionNote?.startsWith(edited + " ")).at(-1); + if (lastAck && field(lastAck.transitionNote!, "revision") === desired.revision) 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, edited, 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, edited, 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 current = this.desired(id, episode, stamp.human, this.opts.queue.transitionLog.listForQitem(id)); + if (!current || this.stopped) return null; + const update = await this.api(UPDATE, { channel: stamp.channel, ts: messageTs, ...handledMessage(message, current.status) }); + if (this.stopped) return null; + if (!update.ok) return this.failed(id, edited, UPDATE, update); + if (update.json.ts !== messageTs || update.json.channel !== stamp.channel) { + return this.failed(id, edited, UPDATE, { ...update, error: "update acknowledgement did not name the requested message" }); + } + const afterUpdate = this.desired(id, episode, stamp.human, this.opts.queue.transitionLog.listForQitem(id)); + if (afterUpdate?.revision !== current.revision) return 0; // The write landed for an obsolete state; reconcile it again. + this.failures.delete(edited); + this.opts.queue.update({ qitemId: id, actorSession: "daemon@kernel", transitionNote: `${edited} revision=${current.revision}` }); + 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-answer-after-question.test.ts b/packages/daemon/test/slack-answer-after-question.test.ts index 2c3ca94f4..be0de8ee7 100644 --- a/packages/daemon/test/slack-answer-after-question.test.ts +++ b/packages/daemon/test/slack-answer-after-question.test.ts @@ -56,6 +56,8 @@ describe("#822: replies answer only the current posted question", () => { inboundMaxConnects: 1, fetchImpl: async (url, init) => { if (url.endsWith("apps.connections.open")) return reply({ ok: true, url: "wss://fake-slack/ws" }); + // This reply-routing fixture keeps post requests, not original Slack blocks. Settlement reads are unavailable. + if (url.includes("conversations.history") || url.includes("conversations.replies")) return reply({ ok: true, messages: [] }); const post = JSON.parse(String(init?.body)); posts.push(post); const overridden = await postHook?.(post); diff --git a/packages/daemon/test/slack-channel-map-wire.test.ts b/packages/daemon/test/slack-channel-map-wire.test.ts index 8af9ae5d0..997ac99d3 100644 --- a/packages/daemon/test/slack-channel-map-wire.test.ts +++ b/packages/daemon/test/slack-channel-map-wire.test.ts @@ -66,6 +66,12 @@ describe("#192 channel map through the real Slack wire", () => { fetchImpl: async (url, init) => { if (url.endsWith("apps.connections.open")) return reply({ ok: true, url: "wss://fake-slack/ws" }); if (url.includes("conversations.history") || url.includes("conversations.replies")) { + // Exact-message settlement is separate from the delivery reconciliation scan counted below. + // This text-only fixture has no retained Block Kit payload; the settlement suite supplies that payload. + const query = new URL(url).searchParams; + if (query.get("limit") === "1" && query.get("inclusive") === "true" && (query.has("latest") || query.has("oldest"))) { + return reply({ ok: true, messages: [] }); + } const raw = String(init?.body ?? ""); const params = raw.startsWith("{") ? JSON.parse(raw) as Record : Object.fromEntries(new URLSearchParams(raw)); const channel = new URL(url).searchParams.get("channel") ?? params.channel ?? ""; diff --git a/packages/daemon/test/slack-reply-to-thread.test.ts b/packages/daemon/test/slack-reply-to-thread.test.ts index f26e98379..01e9a364a 100644 --- a/packages/daemon/test/slack-reply-to-thread.test.ts +++ b/packages/daemon/test/slack-reply-to-thread.test.ts @@ -120,7 +120,11 @@ describe("update --reply-to an earlier item's thread (#96)", () => { const secrets = join(home, "fake.env"); writeFileSync(secrets, "SLACK_BOT_TOKEN=xoxb-EXAMPLE-fake\n"); saveConfig({ ...DEFAULT_CONFIG, enabled: true, channel: "C-TEST", secretsEnvFile: secrets, minimumLevelThatInterrupts: "NOTICE" }, home); posts = []; - wire = buildSlackGatewayWire({ home, queueRepo: repo, registry: { loadHumanRegistry: () => registry, resolveSlackHandle }, fetchImpl: async (_url, init) => { posts.push(JSON.parse(String(init?.body))); return reply({ ok: true, ts: `${posts.length}.1` }); } }); + wire = buildSlackGatewayWire({ home, queueRepo: repo, registry: { loadHumanRegistry: () => registry, resolveSlackHandle }, fetchImpl: async (url, init) => { + // Thread-placement fixtures retain post requests, not the original Block Kit message used by settlement. + if (url.includes("conversations.history") || url.includes("conversations.replies")) return reply({ ok: true, messages: [] }); + posts.push(JSON.parse(String(init?.body))); return reply({ ok: true, ts: `${posts.length}.1` }); + } }); stops.push(() => wire.stop()); wire.startServices?.(); }); @@ -198,6 +202,7 @@ describe("update --reply-to an earlier item's thread (#96)", () => { inboundMaxConnects: 1, fetchImpl: async (url, init) => { if (url.endsWith("apps.connections.open")) return reply({ ok: true, url: "wss://fake-slack/ws" }); + if (url.includes("conversations.history") || url.includes("conversations.replies")) return reply({ ok: true, messages: [] }); const body = JSON.parse(String(init?.body)) as Record; posts.push(body); const ts = `${posts.length}.1`; @@ -793,7 +798,11 @@ describe("update --reply-to an earlier item's thread (#96)", () => { await deliver(first.qitemId); wire.stop(); saveConfig({ ...DEFAULT_CONFIG, enabled: true, channel: "C-OTHER", secretsEnvFile: join(home, "fake.env"), minimumLevelThatInterrupts: "NOTICE" }, home); - wire = buildSlackGatewayWire({ home, queueRepo: repo, registry: { loadHumanRegistry: () => registry, resolveSlackHandle }, fetchImpl: async (_url, init) => { posts.push(JSON.parse(String(init?.body))); return reply({ ok: true, ts: `${posts.length}.1` }); } }); + wire = buildSlackGatewayWire({ home, queueRepo: repo, registry: { loadHumanRegistry: () => registry, resolveSlackHandle }, fetchImpl: async (url, init) => { + // Thread-placement fixtures retain post requests, not the original Block Kit message used by settlement. + if (url.includes("conversations.history") || url.includes("conversations.replies")) return reply({ ok: true, messages: [] }); + posts.push(JSON.parse(String(init?.body))); return reply({ ok: true, ts: `${posts.length}.1` }); + } }); stops.push(() => wire.stop()); wire.startServices?.(); const update = await repo.create({ ...request, humanIntent: "update", replyTo: first.qitemId }); await deliver(update.qitemId); diff --git a/packages/daemon/test/slack-settled-ask.test.ts b/packages/daemon/test/slack-settled-ask.test.ts new file mode 100644 index 000000000..ef15ae24f --- /dev/null +++ b/packages/daemon/test/slack-settled-ask.test.ts @@ -0,0 +1,397 @@ +// Real queue transitions and gateway; Slack HTTP is an in-memory conversation. +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createDb } from "../src/db/connection.js"; +import { migrate } from "../src/db/migrate.js"; +import { ALL_MIGRATIONS } from "../src/db/all-migrations.js"; +import { EventBus } from "../src/domain/event-bus.js"; +import { QueueRepository } from "../src/domain/queue-repository.js"; +import { buildSlackGatewayWire } from "../src/domain/gateway/slack/slack-subsystem.js"; +import { makeQueuePorts } from "../src/domain/gateway/slack/queue-access.js"; +import { DEFAULT_CONFIG, saveConfig } from "../src/domain/gateway/slack/config.js"; +import { resolveSlackHandle } from "../src/domain/gateway/human-registry.js"; + +const human = "person@external"; +const registry = { ok: true as const, entities: [{ entityId: "person", class: "human" as const, + displayName: "Person", address: human, connectorBindings: [{ kind: "slack" as const, + connectorRef: "primary", secretsRef: "env:SLACK_BOT_TOKEN", role: "primary" as const, handle: "UPERSON" }], + prefs: { deliveryClass: "A" as const } }] }; +const response = (body: unknown) => new Response(JSON.stringify(body), { status: 200 }); +type Message = { channel: string; ts: string; text: string; blocks: Array>; thread_ts?: string }; + +describe("posted asks follow settlement outside Slack", () => { + let home: string; + let db: ReturnType; + let repo: QueueRepository; + let wire: ReturnType; + let messages: Message[]; + let updates: Array>; + let postCount: number; + let apiHook: ((method: string, body: Record) => Response | undefined | Promise) | undefined; + let methods: string[]; + let requests: Array<{ url: string; method: string; body: BodyInit | null | undefined; headers: Record }>; + const logs: string[] = []; + + function start() { + wire = buildSlackGatewayWire({ home, queueRepo: repo, + registry: { loadHumanRegistry: () => registry, resolveSlackHandle }, log: (m) => logs.push(m), + 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); + requests.push({ url, method: init?.method ?? "GET", body: init?.body, headers: init?.headers as Record }); + // Tolerate the old read shape here to isolate reopen/blocker/fairness bugs. The request-shape cell below + // independently requires GET, query arguments and no body; this tolerance is not the Slack contract. + const params = { ...Object.fromEntries(api.searchParams), ...body }; + const override = await apiHook?.(method, params); + if (override) return override; + if (api.pathname.endsWith("chat.postMessage")) { + postCount++; + const message = { ...body, ts: `${postCount}.000001` } as Message; + messages.push(message); + return response({ ok: true, ts: message.ts }); + } + if (api.pathname.endsWith("chat.update")) { + updates.push(body); + const message = messages.find((m) => m.ts === body.ts && m.channel === body.channel)!; + Object.assign(message, body); + return response({ ok: true, ts: body.ts, channel: body.channel }); + } + if (api.pathname.endsWith("conversations.history") || api.pathname.endsWith("conversations.replies")) { + const ts = params.latest ?? params.oldest ?? params.ts; + return response({ ok: true, messages: messages.filter((m) => m.channel === params.channel && (!ts || m.ts === ts)) }); + } + return response({ ok: true, messages: [] }); + }, + }); + wire.startServices?.(); + } + + beforeEach(() => { + home = mkdtempSync(join(tmpdir(), "slack-settled-")); + db = createDb(); migrate(db, ALL_MIGRATIONS); + repo = new QueueRepository(db, new EventBus(db), { loadHumanRegistry: () => registry }); + 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; methods = []; requests = []; apiHook = undefined; + start(); + }); + 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(`${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", ...(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)); + const edited = updates[0]!; + expect(edited).toMatchObject({ channel: original.channel, ts: original.ts }); + expect(edited.text).toContain(state === "done" ? "Done" : "Canceled"); + expect(edited.text).toContain("superseded"); + expect(edited.text).toContain(closed.ts); + expect(edited.text).toContain("author@rig"); + expect(edited.text).toContain("Handled in the pane after the merge."); + expect(edited.text).toContain(original.text.slice(original.text.lastIndexOf("\n") + 1)); + const keptText = original.text.split("\n").filter((line) => !line.startsWith("Or reply in this thread") && !line.startsWith("Reply in this thread after this question")).join("\n"); + expect(String(edited.text).endsWith(keptText)).toBe(true); + const blocks = edited.blocks as Array>; + expect(blocks.some((b) => b.type === "actions")).toBe(false); + expect(JSON.stringify(blocks)).not.toMatch(/reply in this thread/i); + for (const block of original.blocks.filter((b) => b.type !== "actions" && b.type !== "context")) expect(blocks).toContainEqual(block); + expect(postCount).toBe(1); + wire.stop(); start(); + await new Promise((r) => setTimeout(r, 25)); + expect(updates).toHaveLength(1); + expect(postCount).toBe(1); + }); + + it("settles only the primary of a multipart ask", async () => { + const item = await postAsk("Supplemental evidence, unchanged."); + expect(messages).toHaveLength(2); + const supplement = structuredClone(messages[1]!); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" }); + await vi.waitFor(() => expect(updates).toHaveLength(1)); + expect(updates[0]!.ts).toBe("1.000001"); + expect(messages[1]).toEqual(supplement); + expect(postCount).toBe(2); + }); + + it("recovers a closure while the wire was stopped from the existing posted receipt and stamp", async () => { + const item = await postAsk(); + wire.stop(); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" }); + expect(updates).toHaveLength(0); + start(); + await vi.waitFor(() => expect(updates).toHaveLength(1)); + expect(updates[0]!.ts).toBe("1.000001"); + expect(postCount).toBe(1); + }); + + it("keeps an unanswered open ask and ordinary progress notes interactive", async () => { + const item = await postAsk(); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", transitionNote: "Still waiting for the person's choice." }); + await new Promise((r) => setTimeout(r, 25)); + expect(updates).toHaveLength(0); + 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: 0, 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); + }); + + it("reads the original with GET query arguments and no body, then writes chat.update as JSON", async () => { + const item = await postAsk(); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" }); + await vi.waitFor(() => expect(updates).toHaveLength(1)); + const read = requests.find((r) => new URL(r.url).pathname.endsWith("conversations.history"))!; + expect(read.method).toBe("GET"); + expect(read.body).toBeUndefined(); + expect(read.headers["content-type"]).toBeUndefined(); + expect(new URL(read.url).searchParams.get("channel")).toBe("C-ASKS"); + expect(new URL(read.url).searchParams.get("latest")).toBe("1.000001"); + expect(new URL(read.url).searchParams.get("inclusive")).toBe("true"); + const write = requests.find((r) => new URL(r.url).pathname.endsWith("chat.update"))!; + expect(write.method).toBe("POST"); + expect(write.headers["content-type"]).toContain("application/json"); + expect(JSON.parse(String(write.body))).toMatchObject({ channel: "C-ASKS", ts: "1.000001" }); + }); + + it("makes the same direct ask visibly open again after close/edit/reopen, without a new post", async () => { + const item = await postAsk(); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" }); + await vi.waitFor(() => expect(updates).toHaveLength(1)); + expect(messages[0]!.text).toMatch(/^Done/); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "pending", reopen: true, transitionNote: "The decision is needed again." }); + expect(repo.getById(item.qitemId)?.state).toBe("pending"); + await vi.waitFor(() => expect(messages[0]!.text).toMatch(/^Open again/)); + expect(messages[0]!.text).toContain("Reply in this thread to answer"); + expect(postCount).toBe(1); + wire.stop(); start(); + await new Promise((r) => setTimeout(r, 25)); + expect(messages[0]!.text).toMatch(/^Open again/); + expect(updates).toHaveLength(2); + }); + + it("keeps the earlier answer on a reopened answered ask without inviting a discarded thread reply", async () => { + const item = await postAsk(); + const input = { sourceSession: human, destinationSession: item.sourceSession, summary: "Answer", body: "Ship it", + inboundReplyTo: item.qitemId, gatewayInbound: true, nudge: false }; + const answer = await repo.createHumanAnswer(input); + expect(repo.getHumanAnswer(item.qitemId, human)?.qitemId).toBe(answer.qitemId); + expect(repo.resolveDirectHumanReply({ qitemId: item.qitemId, actorSession: human, decision: "Ship it" })).toBe(true); + await vi.waitFor(() => expect(updates).toHaveLength(1)); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "pending", reopen: true, transitionNote: "Revisit the decision." }); + expect(repo.getById(item.qitemId)?.state).toBe("pending"); + const repeated = await repo.createHumanAnswer({ ...input, body: "A different answer" }); + expect(repeated.qitemId).toBe(answer.qitemId); + expect(repeated.body).toBe("Ship it"); + await vi.waitFor(() => expect(updates).toHaveLength(2)); + expect(messages[0]!.text).toMatch(/^Open again/); + expect(messages[0]!.text).not.toContain("Reply in this thread to answer"); + expect(messages[0]!.text).toContain("Earlier answer retained"); + expect(postCount).toBe(1); + }); + + it.each(["read", "update"] as const)("reconciles a reopen during an awaited %s, instead of acknowledging obsolete Done", async (stage) => { + const item = await postAsk(); + let entered = false; + let release!: (response: Response) => void; + const gate = new Promise((resolve) => { release = resolve; }); + apiHook = (method, body) => { + if (entered || method !== (stage === "read" ? "conversations.history" : "chat.update")) return; + entered = true; + if (stage === "update") { Object.assign(messages[0]!, body); updates.push(body); } + return gate; + }; + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" }); + await vi.waitFor(() => expect(entered).toBe(true)); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "pending", reopen: true, transitionNote: "Ask is still needed." }); + expect(repo.getById(item.qitemId)?.state).toBe("pending"); + release(stage === "read" ? response({ ok: true, messages: [structuredClone(messages[0]!)] }) + : response({ ok: true, ts: "1.000001", channel: "C-ASKS" })); + await vi.waitFor(() => expect(messages[0]!.text).toMatch(/^Open again/)); + expect(messages[0]!.text).toContain("Reply in this thread to answer"); + expect(postCount).toBe(1); + }); + + it("settles a person's park moved to a live queue blocker, keeping a same-person note interactive", async () => { + const item = await repo.create({ sourceSession: "author@rig", destinationSession: "worker@rig", summary: "Work", body: "Work", nudge: false }); + repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "blocked", blockedOn: human, + summary: "May I proceed?", evidenceRef: "https://example.com/proof" }); + const [alert] = await makeQueuePorts(repo, { loadHumanRegistry: () => registry }).listHumanAlerts({}); + expect(wire.dispatcher.dispatch("post_message", human, alert)).toMatchObject({ ok: true }); + await vi.waitFor(() => expect(repo.currentHumanRequiredEpisode(item.qitemId)?.receipt?.messageTs).toBe("1.000001")); + repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "blocked", blockedOn: human, transitionNote: "Still waiting for the same person." }); + await new Promise((r) => setTimeout(r, 25)); + expect(updates).toHaveLength(0); + const dependency = await repo.create({ sourceSession: "author@rig", destinationSession: "worker@rig", summary: "Dependency", body: "Dependency", nudge: false }); + repo.update({ qitemId: item.qitemId, actorSession: "worker@rig", state: "blocked", blockedOn: dependency.qitemId, transitionNote: "Person's step is complete; waiting on the dependency." }); + expect(repo.getById(item.qitemId)?.blockedOn).toBe(dependency.qitemId); + await vi.waitFor(() => expect(updates).toHaveLength(1)); + expect(messages[0]!.text).not.toContain("Reply in this thread after this question"); + expect(postCount).toBe(1); + }); + + it("does not let six permanent read failures starve an ordinary primary", async () => { + const bad = [await postAsk(), await postAsk(), await postAsk(), await postAsk(), await postAsk(), await postAsk()]; + const good = await postAsk(); + await new Promise((r) => setTimeout(r, 25)); // every still-open post's observation has drained + vi.useFakeTimers(); + const failedReads: string[] = []; + apiHook = (method, params) => { + if (method !== "conversations.history" || params.latest === "7.000001") return; + failedReads.push(String(params.latest)); + return response({ ok: false, error: "missing_scope" }); + }; + for (const item of [...bad, good]) repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on" }); + expect(repo.getById(good.qitemId)?.state).toBe("done"); + await vi.advanceTimersByTimeAsync(60_000); + expect(updates).toHaveLength(1); + expect(updates[0]!.ts).toBe("7.000001"); + expect(failedReads).toHaveLength(6); + expect(postCount).toBe(7); + }); + + it("shows a later close's new reason after the same direct ask was reopened", async () => { + const item = await postAsk(); + wire.stop(); vi.useFakeTimers(); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "no-follow-on", transitionNote: "Initial completion." }); + start(); await vi.advanceTimersByTimeAsync(0); + expect(updates).toHaveLength(1); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "pending", reopen: true, transitionNote: "Need another decision." }); + await vi.advanceTimersByTimeAsync(0); + expect(messages[0]!.text).toMatch(/^Open again/); + await vi.advanceTimersByTimeAsync(15_000); + repo.update({ qitemId: item.qitemId, actorSession: "author@rig", state: "done", closureReason: "superseded", transitionNote: "The later work replaced this ask." }); + await vi.advanceTimersByTimeAsync(0); + expect(updates).toHaveLength(3); + expect(messages[0]!.text).toMatch(/^Done · superseded/); + expect(messages[0]!.text).toContain("The later work replaced this ask."); + expect(postCount).toBe(1); + }); +});