From 2b62a68d594fca9e3683fd619aec8eba148cdc89 Mon Sep 17 00:00:00 2001 From: Tiberiu Socaci Date: Fri, 6 Nov 2026 21:48:22 +0200 Subject: [PATCH] fix: continue Slack agent after secret server approval Signed-off-by: Tiberiu Socaci --- CHANGELOG.md | 2 + FEATURES.md | 2 + TEST-PLAN.md | 15 ++ src/gateway/secret-host-approvals.js | 28 +++- src/gateway/secret-host-continuation.js | 64 ++++++++ src/platforms/durable-inbox.js | 11 +- src/server.js | 5 +- src/slack/message-pipeline.js | 5 +- test/egress-secret-approval.test.js | 12 ++ test/secret-host-continuation.test.js | 195 ++++++++++++++++++++++++ 10 files changed, 330 insertions(+), 9 deletions(-) create mode 100644 src/gateway/secret-host-continuation.js create mode 100644 test/secret-host-continuation.test.js diff --git a/CHANGELOG.md b/CHANGELOG.md index f7ca434a..d7745f83 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ release gates. Stable promotion and publication require approval of the exact te - Restore the agent directory refresh to 03:00 local time after a spring-forward clock change instead of retaining the transition day’s normalized 04:00 hour. +- Automatically continue the original agent thread after approving a server for a hidden variable; no manual “continue” message is needed. + - Clarify that the network tool's policy is advisory in `/sudo` direct-host threads, with no OS-level egress cutoff. Ordinary proxy-mode container enforcement is unchanged. diff --git a/FEATURES.md b/FEATURES.md index a19a5eb2..be6c05f7 100644 --- a/FEATURES.md +++ b/FEATURES.md @@ -1,5 +1,7 @@ # ChannelGate — Features +- **Continue after variable server approval in Slack.** Approving a hidden variable’s server automatically queues one continuation in the original thread as the original requester. It waits behind active work, honors queued Stop and current access, records usage and deduplicates repeated clicks. Queued wakeups survive restart; interrupted continuations are reported without replaying unknown actions. Denial and stale approvals do not continue. + The documentation handbook includes its branded favicon in standalone builds and website exports. The build validator checks that every handbook page references the shipped SVG asset. diff --git a/TEST-PLAN.md b/TEST-PLAN.md index 02cca922..55f738d9 100644 --- a/TEST-PLAN.md +++ b/TEST-PLAN.md @@ -1,5 +1,20 @@ # ChannelGate — Test Plan +## Automatic continuation after variable server approval + +Automated: `node --test test/secret-host-continuation.test.js test/egress-secret-approval.test.js test/durable-approvals.test.js test/approval-links.test.js`. +Live SECRET-HOST-CONTINUE-01: run separately in Claude and Codex worker fixtures with network ON, +Lean OFF, an approved requester and a disposable hidden HTTPS token for an operator-controlled +public endpoint with no approved hosts. Ask to check the endpoint without printing the token. +After secret-refused, click Allow on server without sending continue. Require one automatic retry +and result in the original thread/session as the original requester, including a different admin +approving an organization token. Repeat with a personal owner approval, while the source turn is +active, after it ends, and with a pending card across restart. Require serialized runs and usage +for every continuation. Duplicate clicks, denial, stale/rotated tokens, unauthorized approval, +revoked requester access and Stop while queued must cause no additional execution. Queued wakeups +survive restart; an interrupted running wakeup produces a notice without replaying unknown effects. +Never expose token values. Live engine acceptance remains separate from automated regression. + ## Network tool description and direct-host policy Run `node --test test/channel-policy-audit.test.js test/network-policy.test.js diff --git a/src/gateway/secret-host-approvals.js b/src/gateway/secret-host-approvals.js index ce74d0a2..c97dd9b0 100644 --- a/src/gateway/secret-host-approvals.js +++ b/src/gateway/secret-host-approvals.js @@ -20,8 +20,24 @@ import { engineHostsFor } from "./egress/engine-hosts.js"; import { hostMatches, normalizeHost } from "./egress/rules.js"; import { isValidRuleHost } from "./egress/catalog-rules.js"; import { liveTurnIn } from "./egress/liveness.js"; +import { getDb, fromJson } from "../db/index.js"; export const SECRET_HOST_ACTION = "secret_host"; + +// Stop/clear must retire accepted wakeups even before the inbox can dispatch them (offline or +// capacity-bound). Running handlers are also aborted through the ordinary per-thread queue. +export function cancelSecretHostContinuations(channelId, threadKey = null) { + const db = getDb(); + const rows = db.prepare("SELECT event_id, data FROM inbound_events WHERE namespace = 'secret-host-continuation' AND conversation_id = ? AND status IN ('queued', 'running', 'interrupted', 'notice_failed')").all(channelId); + const threads = new Set(); + for (const row of rows) { + const action = fromJson(row.data, {}); + if (threadKey && action.threadKey !== threadKey) continue; + db.prepare("UPDATE inbound_events SET status = 'done', data = '{}' WHERE namespace = 'secret-host-continuation' AND event_id = ?").run(row.event_id); + if (action.threadKey) threads.add(action.threadKey); + } + return [...threads]; +} // The approval tier a secret's scope carries (mirrored by gateway-server.js secretCardTier). export function secretHostTier(scope) { return scope === "organization" ? "admin" : scope === "personal" ? "owner" : ""; @@ -123,7 +139,7 @@ export async function requestSecretHostApprovals(refusals, { channelId, slug, ho } // The durable executor (server.js setDurableApprovalExecutor) for a SECRET_HOST_ACTION row. -export async function executeSecretHostApproval(record) { +export async function executeSecretHostApproval(record, { enqueueContinuation } = {}) { const action = record?.action; try { if (record?.status !== "executing" || action?.kind !== SECRET_HOST_ACTION || action.channelId !== record.channelId || action.slug !== record.slug @@ -151,7 +167,15 @@ export async function executeSecretHostApproval(record) { if (!entry) throw new Error(`${action.secretName} no longer exists.`); if ((Number(entry.setAt) || 0) !== action.entrySetAt) throw new Error(`${action.secretName} was changed after this approval was requested. Retry the request to get a fresh card.`); await patchSecretEntry({ scope: action.scope, slug: action.slug, userId: action.ownerId, name: action.secretName, addApprovedHost: host }); - return { ok: true, completed: true, label: "secret server approval", message: `${action.secretName} may now be used on ${host}. Retry the request.` }; + let continuation = "Retry the request."; + if (enqueueContinuation && action.authorId && action.threadKey) { + try { + if (enqueueContinuation(record)) continuation = "The agent will continue automatically in this thread."; + } catch { + continuation = "The server was approved, but automatic continuation could not be queued. Say continue to retry."; + } + } + return { ok: true, completed: true, label: "secret server approval", message: `${action.secretName} may now be used on ${host}. ${continuation}` }; } catch (error) { return { ok: false, error: error.message }; } diff --git a/src/gateway/secret-host-continuation.js b/src/gateway/secret-host-continuation.js new file mode 100644 index 00000000..0113f74c --- /dev/null +++ b/src/gateway/secret-host-continuation.js @@ -0,0 +1,64 @@ +// Approval wakeups are durable, deduplicated by approval ID, and serialized with human turns. +import { createDurableInbox } from "../platforms/durable-inbox.js"; +import { automationTarget, postNotice } from "../platforms/notify.js"; +import { getChannelEntry, getChannelMeta, isAdminPrincipal, isApproved } from "../config/store.js"; +import { isAuthorized } from "./modes.js"; +import { runMessage } from "./run.js"; +import { runQueue } from "../slack/message-lifecycle.js"; +import { deliverResult } from "../slack/deliver.js"; +import { createUsageBank } from "./usage.js"; +import { platformOfConversation } from "../platforms/ids.js"; + +export function createSecretHostContinuation({ slack, runner = runMessage, deliver = deliverResult, bank = (payload) => createUsageBank()(payload), target = (id) => automationTarget(slack, id) } = {}) { + let timer; + const inbox = createDurableInbox({ + namespace: "secret-host-continuation", + isReady: (channelId) => Boolean(target(channelId)), + async handle(action) { + const key = `${action.slug}::${action.threadKey}`; + const handle = { aborted: false, controller: new AbortController(), authorId: action.authorId, background: true }; + await runQueue.acquire(key, handle); + try { + if (handle.aborted) return; + // Approval grants a destination, never the approver's identity or runtime privileges. + const entry = await getChannelEntry(action.channelId); + const meta = await getChannelMeta(action.slug); + if (!entry || entry.slug !== action.slug || !meta || !isAuthorized(meta, action.authorId, Boolean(entry.isDM), { + isAdminUser: await isAdminPrincipal(action.authorId), isApprovedUser: await isApproved(action.authorId), + })) return; + if (handle.aborted) return; + const result = await runner({ + channelId: action.channelId, authorId: action.authorId, threadKey: action.threadKey, + origin: "continuation", signal: handle.controller.signal, + text: `[automatic continuation] Server approval was granted for variable ${action.secretName} on ${action.host}. ` + + "Continue the user's pending task and retry the request that was refused with secret-refused. " + + "Check prior results before repeating any completed actions. Never display the variable's value.", + }); + await bank({ channelId: action.channelId, slug: action.slug, authorId: action.authorId, engine: result.engine, taskKind: "background", result }); + if (handle.aborted) return; + await deliver(target(action.channelId), { channel: action.channelId, threadKey: action.threadKey, result, footer: true }); + } finally { runQueue.release(key, handle); } + }, + async interrupted(action) { + const sent = await postNotice(target(action.channelId), { conversationId: action.channelId, threadKey: action.threadKey, + text: "⚠️ The automatic continuation after server approval was interrupted. Check the previous result before continuing; it was not run again." }); + if (!sent) throw new Error("Continuation interruption notice could not be delivered"); + }, + }); + return { + start() { + inbox.start(); + timer ||= setInterval(() => inbox.wake(), 1000); + timer.unref?.(); + }, + stop() { clearInterval(timer); timer = null; inbox.stop(); }, + enqueue(record) { + const action = record.action; + if (!record.id || !action?.authorId || !action.threadKey) return false; + // Other platforms own different session queues; do not bypass their coordination. + if (platformOfConversation(action.channelId) !== "slack") return false; + inbox.accept({ id: record.id, conversationId: action.channelId, payload: action }); + return true; + }, + }; +} diff --git a/src/platforms/durable-inbox.js b/src/platforms/durable-inbox.js index 58dcfe0e..f9b0affb 100644 --- a/src/platforms/durable-inbox.js +++ b/src/platforms/durable-inbox.js @@ -12,7 +12,7 @@ const RETENTION_MS = 7 * 24 * 60 * 60 * 1000; const CLEANUP_INTERVAL_MS = 60_000; const CLEANUP_BATCH = 500; -export function createDurableInbox({ namespace, handle, interrupted = async () => {}, log = console, now = Date.now, concurrencyGroup = "default", conversationKey = null, queuedNamespace = null } = {}) { +export function createDurableInbox({ namespace, handle, interrupted = async () => {}, log = console, now = Date.now, concurrencyGroup = "default", conversationKey = null, queuedNamespace = null, isReady = null } = {}) { if (!namespace || typeof handle !== "function") throw new TypeError("Inbox requires a namespace and handler"); const owner = randomUUID(); // Ordinary work shares four slots. Explicit control intake can use a separate bounded group @@ -50,10 +50,12 @@ export function createDurableInbox({ namespace, handle, interrupted = async () = function pump() { if (stopped) return; if (active.size >= MAX_ACTIVE) return; - for (const row of readyRows(MAX_ACTIVE - active.size)) { + for (const row of readyRows(isReady ? MAX_PENDING : MAX_ACTIVE - active.size)) { + if (active.size >= MAX_ACTIVE) break; const key = `${namespace}:${row.event_id}`; const conversation = `${namespace}:${row.conversation_id}`; if (active.has(key) || activeConversations.has(conversation)) continue; + if (isReady && !isReady(row.conversation_id)) continue; const wasInterrupted = row.status === "interrupted"; const claimed = db.prepare("UPDATE inbound_events SET status = 'running', owner = ? WHERE namespace = ? AND event_id = ? AND status = ?") .run(owner, namespace, row.event_id, row.status); @@ -62,11 +64,11 @@ export function createDurableInbox({ namespace, handle, interrupted = async () = active.add(key); activeConversations.add(conversation); Promise.resolve().then(() => wasInterrupted ? interrupted(fromJson(saved.data, {})) : handle(fromJson(saved.data, {}))) - .then(() => db.prepare("UPDATE inbound_events SET status = 'done', data = '{}' WHERE namespace = ? AND event_id = ? AND owner = ?").run(namespace, row.event_id, owner)) + .then(() => db.prepare("UPDATE inbound_events SET status = 'done', data = '{}' WHERE namespace = ? AND event_id = ? AND owner = ? AND status = 'running'").run(namespace, row.event_id, owner)) .catch((error) => { // A handler may fail AFTER side effects. Keep an explicit interruption record, never // silently turn it back into queued execution. A failed notice waits for reconnect. - db.prepare("UPDATE inbound_events SET status = ? WHERE namespace = ? AND event_id = ? AND owner = ?") + db.prepare("UPDATE inbound_events SET status = ? WHERE namespace = ? AND event_id = ? AND owner = ? AND status = 'running'") .run(wasInterrupted ? "notice_failed" : "interrupted", namespace, row.event_id, owner); log.error?.(`[inbox] ${namespace} event interrupted: ${error?.message || error}`); }) @@ -79,6 +81,7 @@ export function createDurableInbox({ namespace, handle, interrupted = async () = } return { + wake: () => queueMicrotask(pump), accept({ id, conversationId, payload }) { if (stopped) throw new Error("Inbound transport is stopped; event was not accepted"); if (!id || !conversationId) throw new Error("Inbound event requires stable event and conversation identifiers"); diff --git a/src/server.js b/src/server.js index f4eaee27..1461c09e 100644 --- a/src/server.js +++ b/src/server.js @@ -33,6 +33,7 @@ import { BackgroundJobs, setActiveBackgroundJobs } from "./gateway/background.js import { requestApproval, setDurableApprovalExecutor } from "./slack/approvals.js"; import { executeInstructionApproval, INSTRUCTION_ACTION } from "./gateway/instruction-approvals.js"; import { executeSecretHostApproval, SECRET_HOST_ACTION, setSecretHostApprovalRequester } from "./gateway/secret-host-approvals.js"; +import { createSecretHostContinuation } from "./gateway/secret-host-continuation.js"; import { CONTROL_PLANE_ACTION, executeControlPlaneApproval, setControlPlaneInvoker } from "./gateway/control-plane-approvals.js"; import { createDirectDaemonIpc, createGatewayMcpServer, ctxFromClaims } from "./mcp/gateway-server.js"; import { startMcpSocketServer, stopMcpSocketServer, mcpSocketStatus } from "./mcp/socket-server.js"; @@ -339,10 +340,12 @@ async function main() { // job rows have been re-tracked, or the first API/approval-triggered job would rewrite bg_jobs // from a still-empty map and wipe every unrecovered row. backgroundJobs.armRecovery(); + const secretHostContinuation = createSecretHostContinuation({ slack }); + secretHostContinuation.start(); setDurableApprovalExecutor((record) => record.action?.kind === INSTRUCTION_ACTION ? executeInstructionApproval(record) : record.action?.kind === SECRET_HOST_ACTION - ? executeSecretHostApproval(record) + ? executeSecretHostApproval(record, { enqueueContinuation: (saved) => secretHostContinuation.enqueue(saved) }) : record.action?.kind === CONTROL_PLANE_ACTION ? executeControlPlaneApproval(record) : backgroundJobs.startApproved(record)); diff --git a/src/slack/message-pipeline.js b/src/slack/message-pipeline.js index 2289825b..31e452db 100644 --- a/src/slack/message-pipeline.js +++ b/src/slack/message-pipeline.js @@ -73,6 +73,7 @@ import { codeChannelReplyTarget, isCodeChannelMeta, syncCodeChannelMeta } from " import { refreshCodeChannelDiff } from "./code-channels.js"; import { busyThreadChoiceBlocks, busyThreadChoices, deliverBusyThreadChoiceLinks, steerActiveRun, BUSY_THREAD_CHOICE_KIND } from "./busy-thread-choice.js"; import { engineSwitchChoices, engineSwitchChoiceBlocks, engineSwitchChoiceText } from "./engine-switch-choice.js"; +import { cancelSecretHostContinuations } from "../gateway/secret-host-approvals.js"; // In-flight runs by "::". The queue serializes turns per thread — a second // message in the same thread waits FIFO behind the running one instead of racing it in the same @@ -105,7 +106,7 @@ export function abortRunsInChannel(channelId, slug, byUser, threadKey = null) { // durably, so an old button cannot resurrect that message after the active turn was cancelled. const pendingChoices = clearPendingRunChoices({ channelId, threadKey }); const pendingQuestions = cancelPendingQuestions({ channelId, threadKey }); - const stoppedRuns = []; + const stoppedRuns = cancelSecretHostContinuations(channelId, threadKey); const stoppedTurns = []; // Stop every matching engine and terminalize its durable work before the first awaited Slack @@ -132,7 +133,7 @@ export function abortRunsInChannel(channelId, slug, byUser, threadKey = null) { } res.active?.controller?.abort(); // kills a cold (keep-alive off) child mid-run — Claude or Codex abortPooled(key); // terminates a warm (Claude) session - stoppedRuns.push(key.slice(prefix.length)); + if (!stoppedRuns.includes(key.slice(prefix.length))) stoppedRuns.push(key.slice(prefix.length)); for (const [state, handles] of [["active", [res.active]], ["queued", res.queued || []]]) { for (const handle of handles) { if (!handle || handle.stopAccounted) continue; diff --git a/test/egress-secret-approval.test.js b/test/egress-secret-approval.test.js index abf07ee3..f2214638 100644 --- a/test/egress-secret-approval.test.js +++ b/test/egress-secret-approval.test.js @@ -218,6 +218,18 @@ test("executing an approval records the server on the secret itself — for an a // A card for an OLDER version of the secret (deleted and re-created, or re-set) blesses nothing. const stale = await approvals.executeSecretHostApproval({ ...record("U_APPR_ADMIN"), action: { ...action, entrySetAt: 1 } }); assert.match(stale.error, /changed after this approval was requested/); + let wakeups = 0; + const queue = (saved) => { wakeups++; assert.equal(saved.action.authorId, "U_APPR_MEMBER"); return true; }; + const withSource = { ...action, authorId: "U_APPR_MEMBER", threadKey: "1790000000.000100" }; + const denied = await approvals.executeSecretHostApproval({ ...record("U_APPR_MEMBER"), action: withSource }, { enqueueContinuation: queue }); + assert.equal(denied.ok, false); + assert.equal(wakeups, 0); + const resumed = await approvals.executeSecretHostApproval({ ...record("U_APPR_ADMIN"), action: withSource }, { enqueueContinuation: queue }); + assert.equal(wakeups, 1); + assert.match(resumed.message, /continue automatically/); + const queueFailed = await approvals.executeSecretHostApproval({ ...record("U_APPR_ADMIN"), action: withSource }, { enqueueContinuation: () => { throw new Error("full"); } }); + assert.equal(queueFailed.ok, true, "the approved host remains saved even when wakeup fails"); + assert.match(queueFailed.message, /could not be queued/); const ok = await approvals.executeSecretHostApproval(record("U_APPR_ADMIN")); assert.equal(ok.ok, true, ok.error); assert.deepEqual(scoped.getOrgEnv().PAY_API_TOKEN.approvedHosts, ["api.pay.example"]); diff --git a/test/secret-host-continuation.test.js b/test/secret-host-continuation.test.js new file mode 100644 index 00000000..331bbd0e --- /dev/null +++ b/test/secret-host-continuation.test.js @@ -0,0 +1,195 @@ +import test from "node:test"; +import assert from "node:assert/strict"; +import { setTimeout as delay } from "node:timers/promises"; +import { ensureTestEnv } from "./helpers.js"; +ensureTestEnv(); +const { createSecretHostContinuation } = await import("../src/gateway/secret-host-continuation.js"); +const { runQueue } = await import("../src/slack/message-lifecycle.js"); +const store = await import("../src/config/store.js"); +const { getDb } = await import("../src/db/index.js"); +let sequence = 0; +async function fixture() { + const channelId = `C_WAKE_${++sequence}`; + const authorId = `U_WAKE_${sequence}`; + await store.setUser(authorId, { approved: true }); + const entry = await store.upsertChannelEntry(channelId, { name: `wake-${sequence}`, type: "channel", platform: "slack" }); + await store.saveChannelMeta(entry.slug, { ...store.defaultChannelMeta({ channelId, name: entry.name }), access: "open" }); + const action = { kind: "secret_host", channelId, slug: entry.slug, authorId, threadKey: `7000.${sequence}`, secretName: "TEST_TOKEN", host: "api.example.com" }; + return { id: `approval-${sequence}`, action }; +} +async function until(predicate) { + for (let i = 0; i < 200 && !predicate(); i++) await delay(5); + assert.ok(predicate(), "expected asynchronous state"); +} +function status(record) { + return getDb().prepare("SELECT status FROM inbound_events WHERE namespace = 'secret-host-continuation' AND event_id = ?").get(record.id)?.status; +} + +test("approval wakes once, queues behind the source turn, keeps requester and thread, banks every run", async () => { + const record = await fixture(); + const key = `${record.action.slug}::${record.action.threadKey}`; + const source = {}; + await runQueue.acquire(key, source); + const calls = [], delivered = [], banked = []; + const inbox = createSecretHostContinuation({ target: () => ({}), runner: async (args) => { calls.push(args); return { engine: "codex", content: "Retried" }; }, + deliver: async (_, args) => delivered.push(args), bank: async (args) => banked.push(args) }); + inbox.start(); + try { + inbox.enqueue(record); + inbox.enqueue(record); + await until(() => runQueue.count(key) === 2); + assert.equal(calls.length, 0); + runQueue.release(key, source); + await until(() => status(record) === "done"); + assert.equal(calls.length, 1); + assert.equal(calls[0].authorId, record.action.authorId); + assert.equal(calls[0].threadKey, record.action.threadKey); + assert.equal(calls[0].origin, "continuation"); + assert.match(calls[0].text, /retry the request/); + assert.equal(delivered[0].threadKey, record.action.threadKey); + const second = { ...record, id: `${record.id}-second` }; + inbox.enqueue(second); + await until(() => status(second) === "done"); + assert.equal(banked.length, 2); + assert.equal(calls.length, 2); + } finally { runQueue.release(key, source); inbox.stop(); } +}); + +test("stop while queued cancels the wakeup", async () => { + const record = await fixture(); + const key = `${record.action.slug}::${record.action.threadKey}`; + const source = {}; + await runQueue.acquire(key, source); + let calls = 0; + const inbox = createSecretHostContinuation({ target: () => ({}), runner: async () => { calls++; } }); + inbox.start(); + try { + inbox.enqueue(record); + await until(() => runQueue.count(key) === 2); + runQueue.abort(key); + runQueue.release(key, source); + await until(() => status(record) === "done"); + assert.equal(calls, 0); + } finally { runQueue.release(key, source); inbox.stop(); } +}); + +test("revoked requester cannot inherit approver access", async () => { + const record = await fixture(); + await store.setUser(record.action.authorId, { approved: false }); + let calls = 0; + const inbox = createSecretHostContinuation({ target: () => ({}), runner: async () => { calls++; } }); + inbox.start(); + try { + inbox.enqueue(record); + await until(() => status(record) === "done"); + assert.equal(calls, 0); + } finally { inbox.stop(); } +}); + +test("queued approval wakeup survives inbox restart", async () => { + const record = await fixture(); + let calls = 0; + const options = { target: () => ({}), runner: async () => { calls++; return { content: "OK" }; }, deliver: async () => {}, bank: async () => {} }; + const first = createSecretHostContinuation(options); + first.start(); + first.enqueue(record); + first.stop(); // stop before the queued pump runs + const recovered = createSecretHostContinuation(options); + recovered.start(); + try { + await until(() => status(record) === "done"); + assert.equal(calls, 1); + recovered.enqueue(record); + await delay(20); + assert.equal(calls, 1); + } finally { recovered.stop(); } +}); + +test("disconnected destination retains queued wakeup until reconnect", async () => { + const record = await fixture(); + let ready = false, calls = 0; + const inbox = createSecretHostContinuation({ target: () => ready ? {} : null, + runner: async () => { calls++; return { content: "OK" }; }, deliver: async () => {}, bank: async () => {} }); + inbox.start(); + try { + inbox.enqueue(record); + await delay(20); + assert.equal(status(record), "queued"); + assert.equal(calls, 0); + ready = true; + inbox.start(); + await until(() => status(record) === "done"); + assert.equal(calls, 1); + } finally { inbox.stop(); } +}); + +test("interrupted running wakeup posts a notice without replay", async () => { + const record = await fixture(); + getDb().prepare("INSERT INTO inbound_events(namespace, event_id, conversation_id, status, created_ms, owner, data) VALUES ('secret-host-continuation', ?, ?, 'running', ?, 'previous-daemon', ?)") + .run(record.id, record.action.channelId, Date.now(), JSON.stringify(record.action)); + let calls = 0; + const notices = []; + const target = { platform: "slack", post: async (args) => { notices.push(args); return { messageId: "notice" }; } }; + const inbox = createSecretHostContinuation({ target: () => target, runner: async () => { calls++; } }); + inbox.start(); + try { + await until(() => status(record) === "done"); + assert.equal(calls, 0); + assert.match(notices[0].text, /not run again/); + } finally { inbox.stop(); } +}); + +test("foreign platform and legacy approvals do not enter Slack coordination", async () => { + const record = await fixture(); + const inbox = createSecretHostContinuation(); + inbox.start(); + try { + assert.equal(inbox.enqueue({ ...record, action: { ...record.action, channelId: "teams:foreign" } }), false); + assert.equal(inbox.enqueue({ ...record, action: { ...record.action, authorId: "" } }), false); + assert.equal(status(record), undefined); + } finally { inbox.stop(); } +}); + +test("stop or clear terminalizes offline accepted wakeups before acknowledging", async () => { + const record = await fixture(); + const other = { ...record, id: `${record.id}-other`, action: { ...record.action, threadKey: "other-thread" } }; + const { abortRunsInChannel } = await import("../src/slack/message-pipeline.js"); + let ready = false, calls = 0; + const inbox = createSecretHostContinuation({ target: () => ready ? {} : null, runner: async () => { calls++; return {}; }, deliver: async () => {}, bank: async () => {} }); + inbox.start(); + try { + inbox.enqueue(record); + inbox.enqueue(other); + const stopped = abortRunsInChannel(record.action.channelId, record.action.slug, record.action.authorId, record.action.threadKey); + assert.deepEqual(stopped.stoppedRuns, [record.action.threadKey]); + assert.equal(status(record), "done"); + assert.equal(status(other), "queued", "thread stop preserves other threads"); + abortRunsInChannel(record.action.channelId, record.action.slug, record.action.authorId); + assert.equal(status(other), "done"); + ready = true; + inbox.start(); + await delay(20); + assert.equal(calls, 0); + } finally { inbox.stop(); } +}); + +test("stop during execution cannot resurrect the wakeup or post a late answer", async () => { + const record = await fixture(); + const { abortRunsInChannel } = await import("../src/slack/message-pipeline.js"); + let started = false, finish; + const waiting = new Promise((resolve) => { finish = resolve; }); + let deliveries = 0; + const inbox = createSecretHostContinuation({ target: () => ({}), runner: async () => { started = true; await waiting; throw new Error("aborted"); }, + deliver: async () => { deliveries++; }, bank: async () => {} }); + inbox.start(); + try { + inbox.enqueue(record); + await until(() => started); + abortRunsInChannel(record.action.channelId, record.action.slug, record.action.authorId, record.action.threadKey); + finish(); + await until(() => runQueue.count(`${record.action.slug}::${record.action.threadKey}`) === 0); + await delay(10); + assert.equal(status(record), "done"); + assert.equal(deliveries, 0); + } finally { finish(); inbox.stop(); } +});