diff --git a/docs/system-specs/modules/messaging.md b/docs/system-specs/modules/messaging.md index 2047e6c0f60..0c26e600994 100644 --- a/docs/system-specs/modules/messaging.md +++ b/docs/system-specs/modules/messaging.md @@ -64,6 +64,7 @@ Slack's transport path is gated behind the `messaging.use_transport` config flag | `messaging/session_trust.py` | The per-session tool-Trust grant store: `is_session_trusted`, `add_trusted_session(key, sessions=)`, `clear_trusted_sessions`. In memory only, so an ad-hoc auto-approve grant dies with the process. The grant has TWO halves and both are load-bearing: the in-memory mapping the driver reads, and the session's approval policy set to `auto`, because a spawned subagent reads its parent's policy and never this mapping. So it is a `key -> SessionManager` MAPPING rather than a set, which is what lets `clear_trusted_sessions` undo the policy half too (back to `""`, the same value the dashboard's untrust toggle writes) without its caller having to hand a manager back. **Every mutation goes through the API**: reaching the container directly is how a revoke came to drop one half and leave subagents trusted, and a mapping has no `.add`, so a half-grant is not expressible either. Named `session_trust`, not `trust`, so it cannot be confused with a connection-admission roster: this grant is about what ONE session's tools may skip, not about which principals may attach. Consumed only through `TurnDriver`'s `auto_approve_session` predicate, which runs BEHIND the keystone, governance and deny-list gates, so a hard DENY still refuses | | `messaging/link.py` | **Layer 3** — session-key namespacing (`session_key`/`canonical_key`/`legacy_key`/`is_legacy_slack_key`) + `ChannelLink` + DM-scope key derivation / `should_rotate_generation`, plus the in-channel `/link` ⇄ `/unlink` pair (`rebind_conversation_location` / `release_conversation_location`) | | `messaging/conversation.py` | `ConversationState` — per-conversation rotating *generation* bookkeeping (advanced by `/new` and idle/daily reset), seeded from the persisted session map | +| `messaging/inbound_spool.py` | The durable spool for an inbound message the SHUTDOWN GATE refused, and its boot-time replay. See [Durable inbound spool](#durable-inbound-spool-inbound_spoolpy) | | `messaging/session_resume.py` | **Layer 3** — the channel-neutral half of dashboard-session resume. `SessionResumeController` consumes a bound `ResumeSurface` (including its exact durable expectation identity, which may be narrower than `ChannelLink.channel_id`) and owns the complete SHOW-PICKER flow (eligibility/search, audit, nonce/TTL/owner/message scoping, and registration only after a successful post) plus the CHOOSE/BIND transaction (history existence, conflict checks around the awaited settlement, durable expectation before success, atomic inbound claim, lost-claim/storage outcomes, dashboard push, and audit). `SessionBinder` owns inbound routing, settlement, and release. Discord, Telegram, and Teams retain only address/identity derivation, widgets/cards, exact wording and display redaction, callback parsing, and channel-local replay. | | `messaging/resume_expectation.py` | The durable conversation-keyed shadow of those bindings, ONE file per channel (`store_filename`), because a Discord channel id and a Teams conversation id are unrelated address spaces | | `teams/service_urls.py` | `ServiceUrlStore` — durable `conversation_id -> serviceUrl` (plus the authorized identity owning each conversation), because the Bot Framework offers no lookup and a lost reference leaves every proactive path with nowhere to send. `forget` drops a route the Connector permanently refuses | @@ -568,6 +569,24 @@ same phase machine exists twice. `test/test_status_reactions.py` pins both: the shared ladder's phases, debounce, stall marks, close-drain and sink-failure tolerance, and Slack's controller beside them. +## Durable inbound spool (`inbound_spool.py`) + +Gateway shutdown runs channel teardown and `SessionManager.close_all()` concurrently, so a message the platform has already accepted can be refused by the `_closing` gate before its turn opens. Before this module the payload was discarded at that `except SessionClosingError` and the user got the channel's generic fault (or, on Slack, nothing). Issue #2217. + +**What it does, and what it deliberately does not.** It records the refused message durably and, on the next start, tells the user in that same conversation that it was never processed, quoting it back so a resend is one tap. It does NOT re-drive the message as a turn. A re-dispatch half was built and removed: replaying an entry as the operator's own turn makes the spool a second INTAKE path into the model, and every authorization the live path applies at intake (peer allow-list, Telegram's forum gate, Discord's thread roster, WhatsApp's group gate, conversation rotation) has to be re-established on it per channel and kept in step forever. A notice is a proactive SEND, and a proactive send already has exactly one authorization seam, `MessagingTransport.may_send_to`. Scoping the replay to the notice puts the whole feature behind a gate that already exists and is already owned. Re-dispatch, if wanted, is a separate design owned by the channel dispatch wiring (#9144). + +**Written only at the refusal point, never on the happy path.** That scoping is what settles the design questions rather than answering them: there is no ack protocol to design because nothing is written on success; a turn that completed just before exit was never written, so nobody is told to resend something that was answered; the happy path costs zero writes; and platform redelivery is not needed because the pass reads our own disk (nine of ten channels ack before the turn runs anyway). + +**Adoption is opt-in per channel** via `ChannelTurn.inbound_route: InboundRoute`. The route is declared at the channel's dispatch site, where it still holds its normalized envelope, because `ChannelTurn.conversation_id` is a session ATTRIBUTION id (`"weixin:{user}"`), not a reply target. `InboundRoute.text` is the message the USER sent, never `ChannelTurn.user_text` — WhatsApp's rules mode prepends the group's private operating rules to the model prompt, and the restart notice quotes the entry verbatim. A channel that transforms its prompt MUST set it, and there is no fallback to the turn's prompt (an earlier fallback is how a media-only rules-mode message came to spool the group's rules). Weixin captures `text` and the attachment count BEFORE ingestion, because ingestion rewrites the text with turn-owned temp paths and clears `inbound.attachments`; WhatsApp does the same in `receive` via the `pending_original` side table (keyed like `pending_verdicts`), and its dispatcher declares NO route when the entry is absent rather than falling back to the ingested `inbound.text`. **A route is declared only where `may_send_to` can express revocation for it**: Discord's answers from `_allowed_threads`; WhatsApp's answers from `dm_policy` alone and knows nothing of the group roster, so WhatsApp spools DMs only and a refused group message degrades exactly as before this seam. + +**The store.** One JSONL file, `/inbound-spool/refused.jsonl`. A COUNT cap (`SPOOL_MAX_ENTRIES`, newest wins) and an AGE horizon (`SPOOL_MAX_AGE_SECS`) in one primitive — nothing else in the tree combined the two; both are applied on write and again on read. Per-entry text cap with visible truncation. Dedupe on the platform `message_id` ONLY: a body digest would collapse two identical messages on a channel with no id, which is data loss (repeating yourself is ordinary), so an identity-less entry is appended, never matched. The refusal write runs off-loop in `asyncio.to_thread` (it takes a file lock and does disk I/O, and a replay worker may hold the lock) and is wrapped in `asyncio.shield`: the handler that reached the refusal is a task `close_all` is about to cancel, and a bare `await` there would be a cancellation point that orphans the write. With the shield the caller is cancelled and the write is not. An executor already shut down (`RuntimeError`) falls back to an inline write. The whole read-modify-write is serialized by `platform_compat.file_lock` on a dedicated lock file (the spool itself is replaced by rename, so a lock on the old inode would not exclude a writer that opened the new one). **Not for a restricted session**: Telegram and Discord ask their own `_session_restricted(session_key)` (the same predicate that gates the durable-history write) at the refusal point and skip the spool for an incognito or temporary conversation, which promised to persist nothing; the message degrades to the pre-feature loss. A read failure raises `SpoolUnreadable` rather than reading as empty, because every writer rewrites from what it read and the reader unlinks an empty file. Never raises to the caller: a write failure at shutdown degrades to the pre-feature loss; a read failure at boot leaves the file for the next start. + +**The notice pass is AT-LEAST-ONCE, one entry at a time.** `replay_spooled` runs as a detached boot task after the transports are up (`GatewayOrchestrator._replay_spooled_inbound`; `_shutdown` cancels it with a one-second budget). For each entry, oldest first: `peek_next` returns it WITHOUT removing it; the `channels` governance ceiling is asked through `vet_and_audit("channels", channel_type, tool_name="inbound_spool.notice", fail_closed=True)`, the same audited seam every other proactive-send site uses, and a denied channel is HELD (not dropped: the route is not revoked, the channel is governed off, and the horizon bounds it); `may_send_to(conversation_id, thread_id, principal=)` is re-decided (a spooled entry is not a standing grant; a transport with no gate, or one that raises, is read as revoked). **The principal is passed for a DM route only**: a threaded route (Discord thread, Telegram Topic) is authorized by the thread roster alone, because Discord's `may_send_to` falls from a thread not in `_allowed_threads` to `principal in _allowed` on the assumption that a thread route names no principal, and a spooled thread entry does name one (the sender), so passing it would let a still-allowed sender authorize a notice into a thread revoked while the gateway was down; a revoked route is DROPPED and removed with no notice; otherwise `send_message` posts `RESTART_NOTICE` quoting the entry (through `display_safe_for`, so a quoted broadcast mention cannot fire; sized to `capabilities.max_message_chars` with a VISIBLE truncation mark, because the prefix can push a message that fit on the way in over the cap and a transport that slices and still returns an id would otherwise confirm a silently cut notice) and names any dropped attachments; `remove_entry` runs only AFTER the send is confirmed. Confirmation is `messaging.transport.delivery_confirmed` (the shared predicate: a non-empty message id, or any return at all on a transport whose `capabilities.returns_message_id` is `False`, i.e. WeCom and Feishu, which return `""` on success and raise on failure). An UNCONFIRMED send (a raise, an empty id) leaves the entry on disk for the next start and the loop moves on, so one dead route cannot park the queue. **Once per entry per pass**: an entry the pass attempted and left on disk (unconfirmed, or noticed but `remove_entry` returned `False`) is never handed back to that pass, so an unwritable spool costs one notice per entry rather than `SPOOL_MAX_ENTRIES` per entry; an entry that was removed is not remembered, so an identical id-less twin sharing its `trace_id` is still noticed in the same pass. An entry whose channel is not connected THIS run is never touched (a startup blip is not the operator disabling the channel); the age horizon still bounds it. This direction is safe precisely because the only action is a notice: a crash between the send and the removal costs one repeated line, never a repeated side effect, which is the opposite of the tradeoff a re-dispatch would have to make. Removal is by ONE occurrence of the entry's `trace_id` and is always the atomic replace (never a bare unlink, which fails routinely on Windows under an AV handle and would re-notice the entry on every start). + +**Adopted:** Telegram, Discord (with the thread), Weixin, WhatsApp DMs. Slack, Teams, Webex, WeCom, iMessage, Feishu are not yet adopted, tracked in #8912; attachment re-download in #8911; re-dispatch and WhatsApp group routes in #9144. + +**The spool is a trust boundary** — see `security.md`. + ## Layer 3 — session-key namespacing (`link.py`) Session keys are namespaced as `f"{channel_type}:{conversation_id}"` (`session_key()`) so keys never collide across channels (`SLACK_NAMESPACE = "slack"`). Legacy native-Slack sessions were keyed by the bare `thread_ts`; helpers provide the bidirectional `bare ⇄ slack:` shim consumed by `SessionMap` (`session_map.py` imports `ChannelLink` + `canonical_key`, no import cycle): diff --git a/docs/system-specs/modules/security.md b/docs/system-specs/modules/security.md index 5fa878cfad9..494ddead6c4 100644 --- a/docs/system-specs/modules/security.md +++ b/docs/system-specs/modules/security.md @@ -183,6 +183,7 @@ under `(allow default)`, never an edition-resolved or user-writable executable. - **The cron in-flight markers are fenced because the breaker ACTS on them.** `cron-running` (`cron_inflight.RUNNING_DIR_NAME`) is on `_CREW_SECRET_LEAVES` beside `crons.json` and `cron-history`. The reason is sharper than for the store itself: one marker whose PID matches a cron-surface loop-stall dump is what makes `CronService.start()` park that job, so a marker an agent could write is an unauthorized "pause this job" primitive that routes around both the MCP cron tools and the owner-only HTTP surface, and a marker it could delete disables the breaker for a crash loop that is about to recur — the evidence an automatic state change rests on has to be at least as protected as the state it changes. The whole DIRECTORY, so the claim file (`.loop-stall-breaker`), the attribution record (`.loop-stall-attribution`) and any write temporary are covered by one rule. The cron service and `kirocrew doctor` open it directly rather than through this gate, so both keep working, and nothing legitimate reads a marker through a file tool. Beneath the fence `cron_inflight` still refuses what it did not write — a linked (symlink or junction) `cron-running` is neither read from nor written to, children open `O_NOFOLLOW`, single-linked regular files only, size-bounded reads, and `atomic_write(restrict_to_owner=True)` for every write — so a leaf planted before the fence existed is not followed either, and a `read_text` can never block the worker `start()` awaits. - **Off-loop scan in `_resolve_permission`** (the funnel every streamed permission request on cron, Slack, dashboard side-panel and workflow turns goes through): the always-enforced title tier (`is_sensitive_path`, `is_sensitive_bash_command`, `is_denied` on the title) and the tool_input tier (`_first_tool_input_denial` over every string in the payload) run in ONE `asyncio.to_thread` hop, title first, so a request denied on its title reports the title-tier reason and the `always_deny` mechanism, and a request denied on a payload string reports `always_deny_input`. CPython's `re` HOLDS the GIL for the whole of one match call (measured with a tick-counting probe whose clock starts before the worker does: a 5–8 s `search` on a worker leaves the main thread a single tick on 3.10 and 3.12, the same shape as `sorted()` on a large list, while `zlib.compress` — which does release — leaves it ticking), so the hop does NOT keep the loop live inside one scan; the liveness guarantee within a scan is the linear patterns plus `MAX_SCANNABLE_COMMAND_CHARS`, and what the hop buys is the realpath I/O inside `is_sensitive_path` (which releases the GIL) and a yield between the tool_input strings. A ~9 KB shell title scanned inline on the loop was the field crash that motivated this; `hooks.on_tool_call` (HOOK_BASED policy, and the other channel dispatchers that call it synchronously) still runs inline and relies on the gate's linear cost and `MAX_SCANNABLE_COMMAND_CHARS` ceiling for its liveness bound. - Sensitive paths: `~/.aws`, `~/.ssh`, `~/.gnupg`, `~/.gpg`, `~/.config/gcloud`, `~/.azure`, `~/.docker/config.json`, `~/.kube/config`, `~/.npmrc`, `~/.pypirc`, `~/.netrc`, `~/.git-credentials`, `~/.kiro/crew/.env`, `~/.kiro/crew/sel_hmac.key`, `~/.kiro/crew/trust`, `~/.kiro/crew/security_events.jsonl`, `~/.kiro/crew/app_admission.json`, `~/.kiro/crew/workflow_library`, `~/.kiro/crew/run` +- **The refused-inbound spool is fenced as an OUTBOUND SOURCE (keystone directory `inbound-spool`).** `messaging/inbound_spool.py` persists a message the shutdown gate refused and, on the next start, posts a restart notice quoting it into the conversation it names (`docs/system-specs/modules/messaging.md`, "Durable inbound spool"). It holds no credential, and that is exactly why it is easy to leave off the floor: each entry names a conversation and carries text the notice quotes VERBATIM, so a file an agent could WRITE is a way to post text of its choosing, as the gateway, into any conversation still authorized for the principal it names — the `may_send_to` recheck narrows that to authorized routes, which is not a boundary — and an entry holds the verbatim text of a message the operator sent, so READ matters as much. Classified as the whole DIRECTORY on `_CREW_SECRET_LEAVES` (agent file tools, every shell form) and masked on `sandbox._CREW_HIDDEN_LEAVES` (spawned commands), for the reason the `whatsapp` and `apps/aws-control/data` entries are: the spool is written by atomic replace through a sibling temp, and the lock file beside it is what serializes two concurrent refusals, so fencing only the final leaf would leave a writable path to the same bytes. The gateway opens all of it directly rather than through either gate, so spooling and the notice pass keep working. The generalizable rule this landed: **a store whose contents are later SENT on the gateway's behalf needs a credential's fences even though it holds none**, plus the egress authorization (`may_send_to`) re-decided at send time rather than trusted from the entry. - **Crew data-home secret/trust-root leaves are covered under EVERY known home prefix.** Since the data home moved from top-level `~/.kirocrew` to `~/.kiro/crew`, each Kiro Crew secret / governance trust-root leaf (`.env`, `browser-cookies.txt`, `playwright-storage-state.json`, `sel_hmac.key`, `trust`, `security_events.jsonl`, `app_admission.json`, `security_policy.json`, `profiles`, `policy_cache`, `admission_policy.json`, `denied_commands.json`, `crons.json`, `cron-history`, `cron-running`, `workflow_library`, `oauth_endpoints.json`, `live_target.json`, `token_signing.key`, `refresh_chains.json`, `.local_secret`, `routing`, `run`) is expanded onto `_SENSITIVE_HOME_DIRS` under each entry of `_CREW_HOME_PREFIXES = (".kiro/crew", ".kirocrew")`. So the same leaf is read+write-blocked in (1) the current home `~/.kiro/crew` and (2) a not-yet-migrated pre-move legacy `~/.kirocrew`. A legacy `~/.kirocrew` no longer auto-migrates; it survives only in these deny lists, so a host that still has one keeps it read+write-blocked indefinitely rather than for the duration of a move. A new secret is added to `_CREW_SECRET_LEAVES` once and is covered in both locations. - **Identity/auth SQLite store (keystone leaves `data.sqlite3` + WAL/SHM/journal sidecars)** — the store holds live bearer tokens, so a read impersonates the user against the model service and a write forges the identity rows. The kiro-cli and amazon-q copies are fenced by DIRECTORY (`identity_stores.fenced_home_dirs()`), which covers each store's sidecars and temporaries for free. The crew data home cannot be fenced that way — reading `config.json` and `sessions.db` there is routine and intended — so the store is named as a leaf on `_CREW_SECRET_LEAVES`, using `identity_stores.AUTH_SQLITE_DB` rather than a fresh literal so the fence cannot drift from the readers that resolve the same store, and the name is fenced before a writer for that location exists (the treatment `agentcore-inbound` gets). The `-wal`/`-shm`/`-journal` sidecars are named beside it because a file leaf matches its exact name only and a sidecar carries the store's credential bytes — `kiro_cli` states the same fact from the other side, that identity rows read as absent when the `-wal` sidecar is missing; `.tmp`/`.lock` publish artifacts in the same parent are already covered by `_KEYSTONE_ARTIFACT_SUFFIXES`. Scoped to the `_CREW_HOME_PREFIXES` entries and deliberately NOT matched by basename: `data.sqlite3` is a generic filename, so a basename rule would refuse an unrelated application database anywhere under the home directory. Every legitimate reader (`kiro_usage_api`, `kiro_cli`, `kiro_prerequisite`) resolves its path through `identity_stores` and opens it directly rather than through this gate, so no reader is affected. A `-name` traversal from an unfenced ancestor (`find ~ -name data.sqlite3`) names no path either half of the gate can match, and that is true of every keystone leaf rather than of this one. It is a stated residual of the bash gate, not a gap in this fence: the store is refused by `is_sensitive_path()` when the traversal's `-exec` actually opens it, which is where a resolved path exists to be checked. - **Meetings owner-edit root.** `apps/meetings/data/edits` is a directory leaf on the same read+write sensitive floor. The minutes editor returns owner-authored text verbatim, so its sidecars may contain private corrections or credential-shaped examples; meetings agents must neither inspect them with `fs_read` nor overwrite them with `fs_write`. Keeping the root outside `meetings//`, then registering the whole directory, makes the existing hook gate the enforcement boundary and also protects atomic-write temporary siblings. The Meetings backend opens these files directly, so save, overlay, revert, and meeting deletion are unaffected. diff --git a/src/kiro_crew/discord/transport_dispatch.py b/src/kiro_crew/discord/transport_dispatch.py index 7fd65f7ec3f..42d7d2c707a 100644 --- a/src/kiro_crew/discord/transport_dispatch.py +++ b/src/kiro_crew/discord/transport_dispatch.py @@ -72,6 +72,7 @@ ) from kiro_crew.messaging.driver import APPROVAL_INTERACTIVE, TurnDriver from kiro_crew.messaging.identity import channel_inbound_permitted, publish_turn_identity +from kiro_crew.messaging.inbound_spool import InboundRoute, spool_refused_turn from kiro_crew.messaging.link import ( ChannelLink, bind_origin_mirror, @@ -902,6 +903,34 @@ def _begin_monitor_turn() -> None: ) if monitor_completion is not None: return MonitorDispatchResult.BUSY + # Durable inbound spool (issue #2217), for a USER message only — the + # monitor branch above returns first. A monitor turn is generated + # work whose own loop re-fires after the restart, so spooling it + # would replay a check the loop is about to run again anyway. + # Discord has no per-message ack and its resume state is in-memory, + # so our own disk is the only thing that can carry this across the + # restart. + # + # NOT for a restricted session: an incognito or temporary conversation + # is a promise that nothing persists, and the spool is a durable file + # holding the message verbatim. The same predicate that gates the + # durable-history write gates this one. + if not await self._session_restricted(session_key): + await spool_refused_turn( + channel_type="discord", + route=InboundRoute( + conversation_id=channel_id, + # ``msg.text``, NOT the local ``text``: by here the latter + # has attachment context appended, whose inlined temp paths + # are gone after a restart. The spool wants what the user + # typed. + text=msg.text, + user_id=user_id, + thread_id=thread_id or "", + message_id=str(getattr(msg, "message_id", "") or ""), + attachments_dropped=len(getattr(msg, "attachments", None) or ()), + ), + ) except Exception: logger.exception("Discord transport_dispatch: error handling message") if monitor_completion is not None: diff --git a/src/kiro_crew/messaging/dispatch.py b/src/kiro_crew/messaging/dispatch.py index b30335d897c..4cfb206fc59 100644 --- a/src/kiro_crew/messaging/dispatch.py +++ b/src/kiro_crew/messaging/dispatch.py @@ -36,6 +36,7 @@ from kiro_crew.hooks import HOOK_REPLY, TOOL_AUTO_APPROVE, TOOL_DENY, event_is_spawn_run from kiro_crew.messaging.driver import DirectiveConsumer, TurnDriver from kiro_crew.messaging.identity import channel_inbound_permitted, publish_turn_identity +from kiro_crew.messaging.inbound_spool import InboundRoute, spool_refused_turn from kiro_crew.messaging.link import ( DM_SCOPE_UNIFIED, ChannelLink, @@ -206,6 +207,21 @@ class ChannelTurn: audit_caller: str = "" """SEL audit caller label; defaults to ``:unknown``.""" + inbound_route: Optional[InboundRoute] = None + """How to reach this conversation if the SHUTDOWN GATE refuses the turn. + + Supplying it opts the channel into the durable inbound spool: a turn refused + by ``_closing`` is written to disk with this route and replayed on the next + gateway start (:mod:`kiro_crew.messaging.inbound_spool`), instead of the + payload being discarded and the user answered with a generic fault. + + It cannot be derived from :attr:`conversation_id`, which is a session + ATTRIBUTION id (``"weixin:{user}"``) rather than a reply target, so a channel + has to declare its own address here -- it is the only place holding the + normalized envelope. ``None`` (the default) means the channel has not adopted + the spool and its refusal path is byte-identical to before. + """ + #: Every spelling a channel accepts for "abort the running turn". The union of #: the per-channel command tables (``/stop`` and ``/cancel`` everywhere, plus @@ -736,6 +752,21 @@ def _bind_origin() -> None: turn.channel_type, session_key, ) + # Durability, at the ONE point where the payload is still in memory and + # the turn is provably unopened (issue #2217). Every other outcome of this + # dispatch — a completed turn, a turn that ran and failed — is already + # recorded somewhere, which is why nothing is spooled on those paths and + # why a replay cannot double-answer. Best-effort by construction: the + # helper never raises, so a full disk degrades to today's loss rather than + # becoming the thing that fails shutdown. + # ``route.text`` and ONLY ``route.text`` -- never ``turn.user_text``. The + # two differ wherever a channel transforms the prompt, and the difference + # is not cosmetic: WhatsApp's rules mode prepends the group's private + # operating rules to the model prompt, so spooling the turn text would + # quote those rules back into the group in the restart notice. A route + # whose text is empty is a media-only entry (or nothing), not a cue to + # reach for the prompt. + await spool_refused_turn(channel_type=turn.channel_type, route=turn.inbound_route) except Exception: logger.exception("%s transport_dispatch: error handling message", turn.channel_type) if _acquired: diff --git a/src/kiro_crew/messaging/inbound_spool.py b/src/kiro_crew/messaging/inbound_spool.py new file mode 100644 index 00000000000..cce4b3400c0 --- /dev/null +++ b/src/kiro_crew/messaging/inbound_spool.py @@ -0,0 +1,1001 @@ +"""Durable spool for inbound messages the SHUTDOWN GATE refused. + +The loss this closes +-------------------- +A channel accepts an inbound message, the platform tells the user it was +delivered, and then ``get_or_create`` refuses the turn because ``close_all()`` +has already set ``_closing``. No turn ever opened, so there is nothing to drain +and nothing to retry: the payload is discarded and the user is answered with the +channel's generic fault notice (or, on Slack, with silence). + +What this module does about it -- and, deliberately, what it does NOT +--------------------------------------------------------------------- +It records the refused message durably and, on the next start, tells the user +in that same conversation that the message was never processed, quoting it back +so a resend is one tap. It does **not** re-drive the message as a turn. + +The re-dispatch half was built and then removed. Replaying a spooled entry as +the operator's own turn makes the spool a second, parallel INTAKE path into the +model, and every authorization the live path applies at intake -- the peer +allow-list, Telegram's forum gate, Discord's thread roster, WhatsApp's group +gate, conversation rotation -- has to be re-established on it, per channel, and +kept in step with the live path forever. Ten review rounds re-derived that +surface one gate at a time. A notice is a proactive SEND, and a proactive send +already has exactly one authorization seam in this codebase: +``MessagingTransport.may_send_to``. Scoping replay to the notice puts the whole +feature behind a gate that already exists and is already owned, instead of +introducing a parallel one. Re-dispatch, if wanted, is a separate design owned +by the channel dispatch wiring -- issue #9144. + +Why the spool is written at the refusal point and nowhere else +-------------------------------------------------------------- +Nothing is written on the happy path, so there is no ack protocol to design; +every entry is a turn provably refused before it opened, so a completed turn can +never be re-answered; the happy path costs zero writes; and platform redelivery +is not required because replay reads our own disk (nine of ten channels ack +before the turn runs anyway). + +Adoption is opt-in per channel via :attr:`ChannelTurn.inbound_route`. The route +is declared at the channel's dispatch site because ``ChannelTurn.conversation_id`` +is a session ATTRIBUTION id, not a reply target. :attr:`InboundRoute.text` is the +message the USER sent -- never the turn's prompt, which on WhatsApp's rules mode +carries the group's private operating rules, and the notice quotes the entry +(display-safe and size-capped, but otherwise as sent). **A route is declared +only where ``may_send_to`` can express +revocation for it**: Discord answers threads from ``_allowed_threads``; WhatsApp's +answers from ``dm_policy`` alone and knows nothing of the group roster, so +WhatsApp spools DMs only and leaves group routes to #9144. + +Bounding, in one primitive +-------------------------- +The tree had no store combining a COUNT cap and an AGE horizon: +``jsonl_util.rotate_jsonl_at`` gives cap-on-append, the spec-builder tombstones +give slice-on-write, and the subagent tombstone sweep gives an age sweep. This +module is that combination, because a wedged gateway that crash-loops through +shutdown would otherwise accumulate a replay storm for the next start: + +* :data:`SPOOL_MAX_ENTRIES` -- newest-wins count cap, applied on every write. +* :data:`SPOOL_MAX_AGE_SECS` -- an entry older than this is dropped on write and + again on read. A message nobody answered for a day is not worth answering. +* :data:`TEXT_CAP` -- per-entry payload cap, so one message cannot be the whole + budget. + +Delivery is AT-LEAST-ONCE, one entry at a time +---------------------------------------------- +Because the only action is a notice -- idempotent from the user's point of view, +a duplicate costs a repeated line, never a repeated side effect -- the entry is +removed from disk only AFTER the send returns a message id (or the transport +declares the conversation revoked). A crash mid-notice therefore re-notices on +the next start rather than losing the message, which is the opposite of the +tradeoff a re-dispatch would have to make. One entry per pass, so a crash costs +at most one duplicate notice and never the entries queued behind it. An entry +whose channel is not connected THIS run is never noticed: it stays on disk for a +start where the channel is back, and the age horizon bounds it. + +Attachments are NOT spooled (issue #2217's fifth question, still open): an +ingested attachment lives in a turn-owned temp path that is gone after a +restart. The entry records how many were dropped and the notice says so. + +Three things this module is trusted with, and why each is enforced rather than +assumed +---------------------------------------------------------------------------- +* **The spool is a TRUST BOUNDARY even without re-dispatch.** An entry holds the + verbatim text of a message the operator sent, and its route decides which + conversation a proactive send lands in. It lives in its own ``inbound-spool`` + directory under the crew home, fenced from agent file tools + (``security._CREW_SECRET_LEAVES``) and masked in every agent sandbox + (``sandbox._CREW_HIDDEN_LEAVES``); a link planted at the directory, the leaf + or the lock is refused before any read, write or unlink; reads open + ``O_NOFOLLOW`` and ``fstat`` for a plain single-linked file. +* **The route is re-authorized at replay via ``may_send_to``**, the transport's + revocation-at-egress decision, with the entry's ``user_id`` as the principal. + A spooled conversation may have been revoked while the gateway was down, and + the spool is not a standing grant. +* **The read-modify-write is serialized by a file lock**, and a read failure + raises rather than reading as empty -- every writer rewrites from what it read + and the reader unlinks an empty file, so a transient EIO would otherwise erase + the queue. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import hashlib +import json +import logging +import os +import stat as _stat +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Collection, Iterable, Mapping + +from kiro_crew.atomic_write import atomic_write +from kiro_crew.config.paths import data_home +from kiro_crew.jsonl_util import bounded_records +from kiro_crew.messaging.renderer import display_safe_for +from kiro_crew.messaging.transport import TransportCapabilities, delivery_confirmed +from kiro_crew.platform.governance_profiles import vet_and_audit +from kiro_crew.platform_compat import file_lock, is_link_or_junction +from kiro_crew.sel import sel + +logger = logging.getLogger(__name__) + +#: The public surface is exactly what the two production callers use: the +#: refusal sites call ``spool_refused_turn`` (with an ``InboundRoute``) and the +#: gateway calls ``replay_spooled``. Everything else -- the record, the report, +#: the store primitives, the error types, the path -- is module-internal, +#: reached by tests through the module rather than advertised as API. +__all__ = [ + "InboundRoute", + "replay_spooled", + "spool_refused_turn", +] + +#: Newest-wins count cap. Sized for "one shutdown's worth of in-flight +#: messages", not for a backlog: the spool is a rescue buffer, not a queue. +SPOOL_MAX_ENTRIES = 128 + +#: An entry older than this is dropped unreplayed. A day-old unanswered message +#: replayed into a conversation the user has moved on from is noise, and a +#: bounded horizon is what stops a gateway that cannot stay up from growing the +#: spool forever. +SPOOL_MAX_AGE_SECS = 24 * 60 * 60 + +#: Per-entry payload cap in characters. One long paste must not consume the +#: whole budget, and the notice quotes the stored text, so truncation is marked +#: in the text rather than done silently. +TEXT_CAP = 16_384 + +_TRUNCATION_MARK = "\n\n[… truncated when spooled during gateway shutdown]" +_QUOTE_TRUNCATION_MARK = "\n> […] (too long to quote in full — the rest is still yours to resend)" + +#: What a user sees when their channel cannot rehydrate its own inbound. Names +#: the real cause (a restart, not a fault) and quotes the message so resending +#: is one tap. Deliberately not the generic "please try again", which is what +#: this issue's investigation found misleading. +RESTART_NOTICE = ( + "⚠️ The gateway was restarting when this arrived, so it was never " + "processed — nothing is wrong with it. Resend it when you are ready:\n\n{quoted}" +) + +_ATTACHMENT_NOTE = "\n\n(Its {count} attachment(s) are not carried over — resend those too.)" + + +@dataclass(frozen=True) +class InboundRoute: + """How to reach the conversation a refused message came from. + + A channel declares this at its dispatch site, where it still holds its own + normalized envelope. It is deliberately NOT derived from the pipeline's + ``ChannelTurn.conversation_id``: that value is a session-attribution id + (``"weixin:{user}"``) and is not addressable by ``send_message``, so reading + it as a reply target would post the restart notice nowhere. + + Every field but ``conversation_id`` is optional, so a channel supplies what + its platform actually has. A channel that supplies no route at all is not + spooled and behaves exactly as it did before this module existed -- adoption + is opt-in per channel rather than a default that silently half-works. + """ + + conversation_id: str + + text: str = "" + """The message the USER sent, before any turn-side transformation. + + Named separately from ``ChannelTurn.user_text`` because those are not the same + string, and using the turn's is a disclosure bug: WhatsApp's rules mode + prepends ``build_silence_contract(verdict.rules)`` -- the group's private + operating rules -- to the model prompt, so spooling that and quoting it in the + restart notice would publish those rules into the group. The channels that + ingest media also inline turn-owned temp paths into the prompt, which are dead + after a restart. + + REQUIRED for a text-bearing message -- there is no fallback to the turn's + prompt (an earlier fallback is how a media-only rules-mode message came to + spool the group's rules). Empty means media-only, or nothing to spool. + """ + + user_id: str = "" + thread_id: str = "" + message_id: str = "" + attachments_dropped: int = 0 + + +@dataclass(frozen=True) +class SpooledInbound: + """One inbound message the shutdown gate refused, with its routing. + + Every field is plain JSON: the entry survives a process boundary, so it may + hold no live object (no renderer, no socket, no temp path). Exactly the + fields the notice needs and nothing recorded "for later": a field nothing + reads is a field nothing tests, and a re-dispatch design (#9144) would own + its own record. + """ + + channel_type: str + conversation_id: str + text: str + user_id: str = "" + thread_id: str = "" + message_id: str = "" + attachments_dropped: int = 0 + spooled_at: float = 0.0 + + @property + def dedupe_key(self) -> str: + """Identity for collapsing a double-spool of the SAME message, or ``""``. + + Only the platform's own message id can carry this. An empty string means + "this entry has no identity", and such an entry is NEVER collapsed against + another -- which is the point rather than a gap. + + A body digest looks like the obvious fallback for a channel with no id and + is a data-loss bug: two identical messages are ordinary (a repeated + "status", a resent "?"), and on a channel with no message id they hash the + same, so one accepted message would be silently discarded and never + replayed. The hazard the fallback would guard -- the same refusal written + twice -- does not exist: the refusal is one ``except`` branch that runs + once per turn. Replay is at-least-once by design (a notice, not a turn), + so this key is a WRITE-side collapse only and never a replay guard. + """ + if not self.message_id: + return "" + return f"{self.channel_type}:{self.conversation_id}:{self.message_id}" + + @property + def trace_id(self) -> str: + """A short label for one entry in a log line or a replay report. + + Distinct from :attr:`dedupe_key`, which is a correctness identity and is + deliberately empty for an entry the platform gave no id: a log line still + needs something to name. Two identical bodies can share a trace id, which + is acceptable for a diagnostic and is exactly why it must not be used to + decide whether one of them is a duplicate. + """ + tail = self.message_id or hashlib.sha256(self.text.encode("utf-8")).hexdigest()[:8] + return f"{self.channel_type}:{self.conversation_id}:{tail}" + + def to_dict(self) -> dict[str, Any]: + return { + "channel_type": self.channel_type, + "conversation_id": self.conversation_id, + "text": self.text, + "user_id": self.user_id, + "thread_id": self.thread_id, + "message_id": self.message_id, + "attachments_dropped": self.attachments_dropped, + "spooled_at": self.spooled_at, + } + + @classmethod + def from_dict(cls, raw: Mapping[str, Any]) -> "SpooledInbound | None": + """Rebuild an entry, or return ``None`` for anything unusable. + + Fails closed on a record that cannot address a conversation: replaying + into an empty ``conversation_id`` would send the notice nowhere, and a + raise here runs inside the boot path. + """ + try: + channel_type = str(raw.get("channel_type") or "") + conversation_id = str(raw.get("conversation_id") or "") + text = str(raw.get("text") or "") + dropped = max(0, int(raw.get("attachments_dropped") or 0)) + # Text OR media: an entry with neither has nothing to notify about. + if not channel_type or not conversation_id or (not text and not dropped): + return None + return cls( + channel_type=channel_type, + conversation_id=conversation_id, + text=text[:TEXT_CAP], + user_id=str(raw.get("user_id") or ""), + thread_id=str(raw.get("thread_id") or ""), + message_id=str(raw.get("message_id") or ""), + attachments_dropped=dropped, + spooled_at=float(raw.get("spooled_at") or 0.0), + ) + except (TypeError, ValueError): + return None + + +@dataclass +class ReplayReport: + """What one replay pass did, so a caller can log a single line.""" + + notified: list[str] = field(default_factory=list) + """Notice confirmed delivered; entry removed.""" + dropped: list[str] = field(default_factory=list) + """Route no longer authorized; entry removed without a notice.""" + held: list[str] = field(default_factory=list) + """Left on disk for a later start because the channel was not connected.""" + unconfirmed: list[str] = field(default_factory=list) + """Notice attempted but not confirmed; entry left on disk for the next start.""" + + @property + def total(self) -> int: + return len(self.notified) + len(self.dropped) + len(self.held) + len(self.unconfirmed) + + def summary(self) -> str: + return ( + f"{self.total} spooled inbound message(s): " + f"{len(self.notified)} answered with a restart notice, " + f"{len(self.dropped)} dropped (route revoked), " + f"{len(self.held)} held (channel not connected), " + f"{len(self.unconfirmed)} unconfirmed (kept for the next start)" + ) + + +async def spool_refused_turn(*, channel_type: str, route: InboundRoute | None) -> bool: + """Spool a turn the shutdown gate refused. The one call every channel makes. + + The text spooled is :attr:`InboundRoute.text` and nothing else -- the message + the USER sent, declared by the channel beside its reply target. There is + deliberately NO separate text argument and NO fallback to the turn's prompt: + an earlier revision fell back to ``ChannelTurn.user_text`` whenever the + route's text was empty, and an empty route text is exactly what a media-only + rules-mode WhatsApp message produces, so the fallback spooled the group's + private rules -- the string the split exists to keep out -- and the notice + quoted them into the group. A channel that declares a route declares its + text; an empty text with attachments is a media-only entry, and an empty text + with none is nothing to spool. + + Returns False without touching disk when *route* is ``None`` (the channel has + not adopted the seam) or when the message carried nothing at all -- no text + and no attachments. A MEDIA-ONLY message is still recorded: it cannot be + re-dispatched (the media is gone), but the user still sent something the + platform marked delivered, and refusing to write it would make an + uncaptioned screenshot the one shape of message this spool silently drops. + The entry carries an empty body and a nonzero ``attachments_dropped``, so + replay routes it straight to the notice, which names the dropped media. + """ + if route is None: + return False + if not route.text.strip() and not route.attachments_dropped: + return False + return await record_refusal( + SpooledInbound( + channel_type=channel_type, + conversation_id=route.conversation_id, + text=route.text, + user_id=route.user_id, + thread_id=route.thread_id, + message_id=route.message_id, + attachments_dropped=route.attachments_dropped, + ) + ) + + +def spool_path() -> Path: + """Where the spool lives. Resolved per call so a HOME override is honoured. + + Its own top-level directory under the crew home rather than a file beside + other state, because the fences that keep an agent out of it are + DIRECTORY-scoped: an atomic write renames a sibling temp into place, so + fencing only the final name would leave a writable path to the same bytes. + The name is specific for the same reason a fence entry has to be -- a generic + ``messaging`` directory would later acquire siblings that have no business + behind a credential-grade fence. + """ + return data_home() / "inbound-spool" / "refused.jsonl" + + +def _lock_path(path: Path) -> Path: + return path.with_suffix(path.suffix + ".lock") + + +#: Absent on Windows (``getattr`` yields 0, the flag is a no-op). There the +#: ``S_ISREG`` + ``st_nlink`` check on the OPENED descriptor carries the refusal, +#: pinned to the inode actually read rather than to a name that could be swapped +#: after the check. Same shape as ``cron_inflight._read_own_file``. +_O_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) +_O_NONBLOCK = getattr(os, "O_NONBLOCK", 0) + + +class SpoolRedirected(OSError): + """The spool directory or one of its files is a link. + + The spool directory is FENCED from agent writes (``security._CREW_SECRET_LEAVES``, + ``sandbox._CREW_HIDDEN_LEAVES``), but a fence only holds from the build that + ships it. A same-UID agent that planted a symlink at ``inbound-spool`` -- or at + the spool leaf, or the lock -- on a build that predates the fence would have + every open in this module resolve inside whatever the link points at, which + the fence never covered, and a JSON record forged there posts a notice on the + next start into a conversation of the forger's choosing, quoting text the + user never sent. So every path this module + touches is refused when it is a link, BEFORE the lock is taken, the file is + read, written or unlinked. Raised as its own ``OSError`` so the best-effort + callers treat it exactly like every other refusal: the message degrades to + the pre-feature loss (on write) or is left for a human (on read), and the + refusal is logged at WARNING because a link here is never accidental. + """ + + +def _refuse_links(path: Path) -> None: + """Refuse when the spool DIRECTORY, the spool LEAF or the LOCK is a link. + + The directory is checked because ``mkdir(exist_ok=True)`` succeeds on a link + to a directory and every child open then follows it; the leaf and the lock + because ``os.open``/``unlink``/``os.replace`` resolve the final component + differently and a link at any of them is a redirect for at least one of + those. lstat-based, so not race-free against a link planted BETWEEN this + check and the open -- the reads below close that window with ``O_NOFOLLOW`` + and an ``fstat`` on the opened descriptor, and the writes go through + ``atomic_write(restrict_to_owner=True)``, whose own linked-parent walk is the + writers' half of the same rule. What this check removes on its own is the + PRE-PLANTED shape, which is the one an attacker can set up at leisure. + """ + for candidate in (path.parent, path, _lock_path(path)): + try: + linked = is_link_or_junction(candidate) + except OSError as exc: + raise SpoolRedirected(f"cannot inspect {candidate}: {exc}") from exc + if linked: + raise SpoolRedirected(f"inbound spool path is a link, refusing: {candidate}") + + +def _open_own_file(path: Path) -> int | None: + """Open *path* for reading only if it is a plain, single-linked regular file. + + Returns the descriptor, or ``None`` when the file is absent. Raises + :class:`SpoolRedirected` for a link, a FIFO, a device or a hard-linked inode -- + each is a way to make this module read bytes it did not write. The type check + runs on the descriptor that is then read (``fstat``), so there is no window in + which the name is swapped between check and use. + """ + try: + fd = os.open(path, os.O_RDONLY | _O_NOFOLLOW | _O_NONBLOCK) + except FileNotFoundError: + return None + try: + st = os.fstat(fd) + except OSError: + os.close(fd) + raise + if not _stat.S_ISREG(st.st_mode) or st.st_nlink > 1: + os.close(fd) + raise SpoolRedirected(f"inbound spool leaf is not a plain single-linked file: {path}") + return fd + + +@contextlib.contextmanager +def _spool_lock(path: Path) -> Any: + """Serialize a whole read-modify-write on the spool. + + Refuses a linked directory, leaf or lock FIRST -- see :func:`_refuse_links`. + + Two inbound messages refused in the same shutdown are two concurrent + ``asyncio.to_thread`` writers, and the boot pass reads and removes in workers + of its own. Without this they read the same snapshot and the second atomic + replace silently drops the first -- reintroducing the loss this module + exists to close, one layer up. + + A DEDICATED lock file, not the spool itself: the spool is replaced by rename, + so a lock held on the old inode would not exclude a writer that opened the + new one. ``file_lock`` fails closed, and this whole module is best-effort, so + a lock that cannot be taken propagates to the caller's ``except`` and the + message degrades to the pre-feature drop rather than to a torn file. + """ + lock = _lock_path(path) + lock.parent.mkdir(parents=True, exist_ok=True) + _refuse_links(path) + # O_NOFOLLOW on the lock too: a link planted between the check above and + # this open would otherwise lock (and create) a file somewhere else. + fd = os.open(lock, os.O_CREAT | os.O_RDWR | _O_NOFOLLOW, 0o600) + try: + with file_lock(fd): + yield + finally: + os.close(fd) + + +def _clamp_text(text: str) -> str: + if len(text) <= TEXT_CAP: + return text + keep = TEXT_CAP - len(_TRUNCATION_MARK) + return text[: max(0, keep)] + _TRUNCATION_MARK + + +class SpoolUnreadable(OSError): + """The spool exists but could not be read. + + Deliberately NOT collapsed into "empty". Both writers rewrite the file from + what they read, and the reader unlinks a file it read as empty -- so + reporting a transient read failure (EIO, a permissions blip, a filesystem + that is remounting) as ``[]`` would have the next write ERASE every queued + message. A caller that sees this leaves the file untouched. + """ + + +def _read_entries(path: Path) -> list[SpooledInbound]: + """Every parseable entry in *path*, oldest first. + + A MISSING file is empty. Any other read failure raises + :class:`SpoolUnreadable` rather than returning ``[]`` -- see that class for + why the distinction is load-bearing here when it usually is not. + """ + entries: list[SpooledInbound] = [] + try: + fd = _open_own_file(path) + if fd is None: + return [] + with os.fdopen(fd, "rb") as handle: + for line in bounded_records(handle, path, label="inbound spool"): + stripped = line.strip() + if not stripped: + continue + try: + raw = json.loads(stripped) + except ValueError: + continue + if not isinstance(raw, dict): + continue + entry = SpooledInbound.from_dict(raw) + if entry is not None: + entries.append(entry) + except SpoolRedirected: + raise + except OSError as exc: + raise SpoolUnreadable(f"inbound spool unreadable at {path}: {exc}") from exc + return entries + + +def _prune(entries: Iterable[SpooledInbound], *, now: float) -> list[SpooledInbound]: + """Apply the age horizon then the count cap (newest wins), oldest first. + + Both bounds are applied on WRITE and again on READ. Applying them twice is + deliberate: a spool written by a build with a looser cap, or one that sat on + disk past the horizon while the gateway was down, must not be replayed just + because it was legal when it was written. + """ + fresh = [e for e in entries if now - e.spooled_at <= SPOOL_MAX_AGE_SECS] + if len(fresh) > SPOOL_MAX_ENTRIES: + fresh = fresh[-SPOOL_MAX_ENTRIES:] + return fresh + + +def _serialize(entries: Iterable[SpooledInbound]) -> str: + return "".join(json.dumps(e.to_dict(), separators=(",", ":")) + "\n" for e in entries) + + +def record_refusal_sync( + entry: SpooledInbound, + *, + path: Path | None = None, + now: float | None = None, +) -> bool: + """Persist *entry*. Returns True when it is on disk, False on any refusal. + + Read-modify-write of the whole file rather than a bare append, because the + count cap and the dedupe both need the existing set. That is affordable + precisely because the file is bounded to :data:`SPOOL_MAX_ENTRIES` lines and + is only ever touched on the refusal path. + + Never raises. This runs while the gateway is already shutting down, so a + full disk or a read-only home must degrade to "the message is lost, as it + was before" rather than becoming the thing that fails shutdown. + """ + target = spool_path() if path is None else path + stamp = time.time() if now is None else now + try: + record = SpooledInbound( + channel_type=entry.channel_type, + conversation_id=entry.conversation_id, + text=_clamp_text(entry.text), + user_id=entry.user_id, + thread_id=entry.thread_id, + message_id=entry.message_id, + attachments_dropped=entry.attachments_dropped, + spooled_at=entry.spooled_at or stamp, + ) + if not record.channel_type or not record.conversation_id: + return False + if not record.text.strip() and not record.attachments_dropped: + return False + target.parent.mkdir(parents=True, exist_ok=True) + # The lock spans the READ and the replace, not just the write: the cap and + # the dedupe are both decided from the existing set, so a snapshot taken + # outside it is stale by the time it is written back and the other + # writer's message is dropped. + with _spool_lock(target): + existing = _read_entries(target) + key = record.dedupe_key + # An identity-less entry (no platform message id) is appended rather + # than matched: collapsing two of those would discard a second + # genuine message whose text happens to be identical. + kept = [e for e in existing if not key or e.dedupe_key != key] + kept.append(record) + # restrict_to_owner: owner-only mode AND atomic_write's own + # linked-parent refusal, the writers' half of _refuse_links. + atomic_write( + target, _serialize(_prune(kept, now=stamp)), newline="", restrict_to_owner=True + ) + return True + except Exception: + logger.warning( + "inbound spool: could not record refused %s message", entry.channel_type, exc_info=True + ) + return False + + +async def record_refusal(entry: SpooledInbound, *, path: Path | None = None) -> bool: + """Off-loop :func:`record_refusal_sync`, shielded from the caller's cancel. + + The write takes a file lock and does disk I/O, so it belongs off the loop: + a replay worker may hold the lock, and blocking the loop on it would stall + every other task in a shutdown that is already racing a deadline. + + But the handler task that reached the refusal is one ``close_all`` is about + to cancel, and a bare ``await asyncio.to_thread(...)`` is a cancellation + point: the cancel would land there, this coroutine would unwind, and the + write would be an orphan nobody awaits. ``asyncio.shield`` is what keeps + those two apart -- the caller may be cancelled, the write is not, and this + coroutine still returns its result when the caller is allowed to finish. + (The residual is the process's ``os._exit`` landing while the worker is + mid-write, which no in-process shape can close and which loses at most the + one message being written.) + + The default executor may already be refusing work at this point + (``RuntimeError: cannot schedule new futures after shutdown``); dropping the + message because the pool closed first would reintroduce the loss this module + exists to close, so that one case falls back to writing inline -- one small + bounded file, on a loop that is being torn down anyway. + """ + try: + return await asyncio.shield(asyncio.to_thread(record_refusal_sync, entry, path=path)) + except asyncio.CancelledError: + raise + except RuntimeError: + return record_refusal_sync(entry, path=path) + except Exception: + logger.warning("inbound spool: off-loop record failed", exc_info=True) + return record_refusal_sync(entry, path=path) + + +def peek_next( + *, + path: Path | None = None, + now: float | None = None, + connected: Collection[str] | None = None, + skip: Collection[str] = (), +) -> tuple[SpooledInbound | None, list[str]]: + """Return ``(oldest actionable entry or None, held trace ids)`` WITHOUT removing. + + An entry is actionable when its channel is in *connected* (``None`` means + every channel) and its ``trace_id`` is not in *skip*. An entry whose channel + is NOT connected is left in place and its trace id is returned in the second + element, so the caller can report it as held without a second read. Stale + entries are pruned here rather than returned, so the age horizon is enforced + on the read as well as on the write -- and an all-stale file is removed so it + is not re-read on every start. + + The entry stays on disk until :func:`remove_entry` is called for it, which the + replay driver does only after the notice is confirmed delivered. That is the + at-least-once half: a crash between the send and the removal re-notices, it + does not lose. + + Never raises. An unreadable or linked spool yields ``(None, [])`` and is left + exactly as it was. + """ + target = spool_path() if path is None else path + stamp = time.time() if now is None else now + held: list[str] = [] + try: + with _spool_lock(target): + entries = _read_entries(target) + remaining = _prune(entries, now=stamp) + if not remaining: + if target.exists(): + with contextlib.suppress(OSError): + target.unlink() + return None, held + if len(remaining) != len(entries): + # Persist the prune so a stale entry is not re-parsed every pass. + atomic_write(target, _serialize(remaining), newline="", restrict_to_owner=True) + for entry in remaining: + if connected is not None and entry.channel_type not in connected: + held.append(entry.trace_id) + continue + if entry.trace_id in skip: + continue + return entry, held + return None, held + except SpoolRedirected: + logger.warning("inbound spool: %s is a link; refusing to read or replay it", target) + return None, held + except SpoolUnreadable: + logger.warning( + "inbound spool: could not read %s; leaving it for the next start", + target, + exc_info=True, + ) + return None, held + except OSError: + logger.warning("inbound spool: could not read %s", target, exc_info=True) + return None, held + + +def remove_entry(entry: SpooledInbound, *, path: Path | None = None) -> bool: + """Remove *entry* from the spool. Called only once its notice is confirmed. + + Matched by identity (``trace_id``), so a second entry with identical text on + an id-less channel is not removed along with it. The removal is the atomic + replace on every branch -- never a bare unlink, which on Windows fails + routinely under an AV scanner's handle and would leave the entry to be + noticed again. The unlink of an emptied spool is a best-effort tidy-up that + gates nothing. + + Never raises; ``False`` means the entry is still on disk and will be noticed + again on the next start, which is the safe direction. + """ + target = spool_path() if path is None else path + try: + with _spool_lock(target): + existing = _read_entries(target) + # trace_id is channel+conversation+id (or a body digest), so two + # identical id-less bodies share one; remove exactly ONE occurrence. + removed = False + rest: list[SpooledInbound] = [] + for e in existing: + if not removed and e.trace_id == entry.trace_id: + removed = True + continue + rest.append(e) + if not removed: + return False + atomic_write(target, _serialize(rest), newline="", restrict_to_owner=True) + if not rest: + with contextlib.suppress(OSError): + target.unlink() + return True + except Exception: + logger.warning( + "inbound spool: could not remove %s entry", entry.channel_type, exc_info=True + ) + return False + + +def _quote(entry: SpooledInbound, transport: Any) -> str: + """The restart notice for *entry*, made display-safe and sized for *transport*. + + The quoted body is the user's own text, but it still passes through the + channel-neutral defang: a message injected into a group conversation can + carry a broadcast mention, and echoing it back unescaped would fire it. The + defang runs on the QUOTE before it is sized, because it inserts characters. + + Sized to ``capabilities.max_message_chars``, because the notice PREFIXES the + quote: a message that fit the platform's cap on the way in no longer fits + with the prefix and the ``> `` markers added, and a transport's ``send_message`` + that slices to its cap and returns an id would confirm a notice whose tail + was silently cut. The quote is truncated here, VISIBLY, before the send, so + what the user sees says so. It is not chunked into several messages: the + quote is an echo of text the sender still holds and is told to resend, so a + marked truncation is an honest notice while a multi-part send with per-chunk + confirmation would be a delivery protocol for an echo. + """ + body = entry.text + if entry.attachments_dropped: + body += _ATTACHMENT_NOTE.format(count=entry.attachments_dropped) + quoted = "\n".join(f"> {line}" for line in body.splitlines() or [""]) + capabilities = getattr(transport, "capabilities", None) + if capabilities is not None: + try: + quoted = display_safe_for(quoted, capabilities) + except Exception: + pass + cap = int(getattr(capabilities, "max_message_chars", 0) or 0) + if cap > 0: + room = cap - len(RESTART_NOTICE.format(quoted="")) + if len(quoted) > room: + keep = max(0, room - len(_QUOTE_TRUNCATION_MARK)) + quoted = quoted[:keep].rstrip() + _QUOTE_TRUNCATION_MARK + return RESTART_NOTICE.format(quoted=quoted) + + +def _audit_route(entry: SpooledInbound, outcome: str) -> None: + """SEL record for the ``may_send_to`` decision on *entry*, grant or denial. + + Both outcomes, as the cross-surface proactive send records them: a denial + also DELETES a stored user message, and a grant is what puts the gateway's + own text into a conversation on the boot path. Best-effort, so a SEL failure + cannot turn either into a raise where an exception costs the gateway its + start. + """ + try: + sel().log_api_access( + caller=entry.user_id or "unknown", + operation="channel.proactive_send_authorize", + outcome=outcome, + source=entry.channel_type, + resources=f"inbound-spool -> {entry.channel_type}:{entry.conversation_id}", + ) + except Exception: + logger.debug("SEL logging failed for inbound-spool authz %s", outcome, exc_info=True) + + +def _route_authorized(entry: SpooledInbound, transport: Any) -> bool: + """Whether the notice may still be posted to this conversation. + + A spooled entry is not a standing grant: the peer may have left the roster, + a Discord thread may have been revoked, all while the gateway was down. + ``may_send_to`` is the transport's own revocation-at-egress decision and the + notice is a proactive send, which is exactly what it governs. Fails CLOSED on + a transport that cannot answer or that raises. + + The principal is passed for a DM route ONLY. A THREADED route (a Discord + thread, a Telegram forum Topic) is authorized by the thread roster and + nothing else: Discord's ``may_send_to`` falls from a thread not in + ``_allowed_threads`` to its DM arm, ``principal in _allowed``, on the stated + assumption that a thread route names no principal. A spooled thread entry + DOES name one -- the sender -- so passing it would let a still-allowed sender + authorize a notice into a thread that was revoked while the gateway was + down. The sender being on the DM allow-list says nothing about whether the + thread may be posted to; withholding the principal keeps the two rosters + from answering for each other. + """ + gate = getattr(transport, "may_send_to", None) + if gate is None: + return False + thread = entry.thread_id or None + principal = "" if thread else entry.user_id + try: + permitted = bool(gate(entry.conversation_id, thread, principal=principal)) + except Exception: + logger.warning( + "inbound spool: %s may_send_to raised; treating the route as revoked", + entry.channel_type, + exc_info=True, + ) + permitted = False + _audit_route(entry, "allowed" if permitted else "denied") + return permitted + + +def _channel_permitted_sync(channel_type: str) -> bool: + """The ``channels`` governance ceiling, audited, for one notice send. + + The notice is a proactive send on a network surface, and every other + proactive-send site in the tree (``channel.send_message`` on the dashboard, + the cron fallback legs) asks this same seam before sending. A policy that + denies the channel is an ordinary operational event -- an operator can tighten + it while the gateway is down -- and the transport being CONNECTED says nothing + about whether it may be written to. Fail-closed: a degraded evaluation denies. + Same seam as :func:`kiro_crew.dashboard.handlers.messaging._vet_channel_send`, + so the audit record has the same shape; the ``tool_name`` names this caller. + """ + try: + decision = vet_and_audit( + "channels", + channel_type, + session_key=f"inbound-spool:{channel_type}", + tool_name="inbound_spool.notice", + fail_closed=True, + ) + return bool(getattr(decision, "permitted", False)) + except Exception: + logger.warning("inbound spool: channel governance check failed", exc_info=True) + return False + + +async def _notify_one(entry: SpooledInbound, transport: Any) -> str: + """Send the notice. Returns ``"notified"`` or ``"dropped"``; raises on unconfirmed. + + Raising is how "keep the entry" is expressed: the caller removes an entry + from disk only when this returns. A ``dropped`` return also removes it -- the + route is revoked, and there is nothing more this module may do with it. + """ + if not _route_authorized(entry, transport): + logger.info( + "inbound spool: dropping %s message — the conversation is no longer an " + "authorized destination", + entry.channel_type, + ) + return "dropped" + send = getattr(transport, "send_message", None) + if send is None: + return "dropped" + message_id = await send( + entry.conversation_id, _quote(entry, transport), entry.thread_id or None + ) + # The shared predicate, not a local re-spelling: it is what knows that WeCom and + # Feishu return "" on SUCCESS (``returns_message_id=False``), and a local copy + # that forgot that would re-notice those two forever. + capabilities = getattr(transport, "capabilities", None) or TransportCapabilities() + if not delivery_confirmed(capabilities, str(message_id or "")): + raise RuntimeError("notice send returned no message id") + return "notified" + + +async def replay_spooled( + *, + transports: Mapping[str, Any], + path: Path | None = None, + now: float | None = None, +) -> ReplayReport: + """Notify every spooled entry's conversation that its message was not processed. + + *transports* is the live ``channel_type -> MessagingTransport`` map the host + already keeps (``DashboardState.channel_transports``). An entry whose channel + is not connected THIS run is left on disk untouched -- a channel can be absent + because its startup transiently failed -- and the age horizon bounds it. + + AT-LEAST-ONCE: an entry is removed only after its notice is confirmed + delivered or its route is found revoked. An unconfirmed send (a raise, an + empty message id) leaves the entry in place for the next start, and the loop + moves on so one bad route cannot block the rest. A crash between the send and + the removal re-notices on the next start; a duplicate notice is a repeated + line, never a repeated side effect, which is why this direction is safe here + and would not have been for a re-dispatch. + + Once per entry per pass. An entry the pass attempted and LEFT ON DISK -- an + unconfirmed send, or a notice that landed but whose removal failed -- is never + returned to it again, so a spool that has become unwritable costs one notice + per entry and not one per loop iteration; the next start tries again. An + entry that was removed is not remembered: two identical id-less bodies share + a ``trace_id``, and forgetting the removed one is what lets its twin be + noticed in the same pass rather than the next. + + Never raises -- this runs on the boot path, where an exception would cost the + gateway its start. + """ + report = ReplayReport() + connected = set(transports) + seen: set[str] = set() + + for _ in range(SPOOL_MAX_ENTRIES): + entry, held = await asyncio.to_thread( + peek_next, path=path, now=now, connected=connected, skip=seen + ) + for trace_id in held: + if trace_id not in seen: + seen.add(trace_id) + report.held.append(trace_id) + logger.info( + "inbound spool: leaving %s message on disk — channel is not connected", + trace_id.split(":", 1)[0], + ) + if entry is None: + break + # The governance ceiling, per entry rather than once per channel: the + # decision is cheap, the pass is bounded, and asking per send is the + # shape every other proactive-send site uses. A denied channel is HELD, + # not dropped: the route is not revoked, the channel is governed off, and + # the operator may loosen the policy before the age horizon expires it. + if not await asyncio.to_thread(_channel_permitted_sync, entry.channel_type): + if entry.trace_id not in seen: + logger.info( + "inbound spool: leaving %s message on disk — channel is denied by the " + "active governance profile", + entry.channel_type, + ) + report.held.append(entry.trace_id) + seen.add(entry.trace_id) + continue + transport = transports[entry.channel_type] + try: + outcome = await _notify_one(entry, transport) + except Exception: + logger.warning( + "inbound spool: notice for %s not confirmed; keeping the entry for the next start", + entry.channel_type, + exc_info=True, + ) + seen.add(entry.trace_id) + report.unconfirmed.append(entry.trace_id) + continue + if not await asyncio.to_thread(remove_entry, entry, path=path): + # The notice landed but the entry could not be taken off disk. Marked + # seen so this pass will not notice it again; the next start will + # (one duplicate line), which is the at-least-once contract. + seen.add(entry.trace_id) + logger.warning( + "inbound spool: notice for %s sent but the entry could not be removed; " + "it will be noticed again on the next start", + entry.channel_type, + ) + report.unconfirmed.append(entry.trace_id) + continue + getattr(report, outcome).append(entry.trace_id) + if report.total: + logger.info("inbound spool: replayed %s", report.summary()) + return report diff --git a/src/kiro_crew/sandbox.py b/src/kiro_crew/sandbox.py index 195acac656e..3395e3de94e 100644 --- a/src/kiro_crew/sandbox.py +++ b/src/kiro_crew/sandbox.py @@ -208,6 +208,16 @@ def _overflow_uid() -> int | None: "aws-control-staging", "apps/meetings/data/edits", "whatsapp", + # The refused-inbound spool. Fenced from agent FILE TOOLS by + # ``security._CREW_SECRET_LEAVES``; masked here so a spawned command cannot + # reach it either -- an entry an agent could write is posted on the next + # start, verbatim, as a gateway-authored notice into the conversation it + # names. Whole directory, because the spool is written by atomic replace via + # a sibling temp name, and because the lock file beside it is what serializes + # concurrent refusals. Nothing inside the sandbox reads or writes it: both + # the spool write and the notice pass happen in the GATEWAY process, which + # opens the paths directly. + "inbound-spool", # The Notes state files below are OWNED by the md-notebook backend, which is itself # a sandboxed spawn (`apps/backend.py`), so the mask alone would break the app: the # registry write's final rename gets EPERM and attach/clone always fails (#8762). diff --git a/src/kiro_crew/security/paths.py b/src/kiro_crew/security/paths.py index 99165316947..9dcf8af2628 100644 --- a/src/kiro_crew/security/paths.py +++ b/src/kiro_crew/security/paths.py @@ -272,6 +272,25 @@ def _leaf_basename(spec: str) -> str: "aws-control-staging", "browser-cookies.txt", "playwright-storage-state.json", + # The refused-inbound spool (messaging/inbound_spool.py). Not a secret: it is + # an OUTBOUND SOURCE. Each entry names a conversation and carries text the + # gateway posts on the next start, verbatim, in a restart notice to that + # conversation -- so a file an agent could write is a way to send text of the + # agent's choosing, as the gateway, into any conversation still authorized + # for the principal it names. The egress recheck (may_send_to) narrows that + # to authorized routes, which is not a boundary. Read matters too: an entry + # holds the verbatim text of a message the operator sent, which is exactly + # the private prompt content the rest of this floor exists to keep + # unreadable. + # + # Classified as the whole DIRECTORY, for the reason the ``whatsapp`` and + # ``apps/aws-control/data`` entries are: the spool is written by atomic + # replace through a sibling temp name in the same directory, so fencing only + # the final leaf would leave a writable path to the same bytes -- and the + # lock file beside it is what serializes two concurrent refusals. The gateway + # opens all of it directly rather than through this gate, so spooling and + # the notice pass keep working. + "inbound-spool", # Per-session work ledgers (session_ledger.py). Not credentials, but each # directory is one session's private work state, and the ledger's whole # authorization model is "a session reaches only its OWN ledger" (the HTTP diff --git a/src/kiro_crew/slack/gateway.py b/src/kiro_crew/slack/gateway.py index cf675c086f0..aec9d91c526 100644 --- a/src/kiro_crew/slack/gateway.py +++ b/src/kiro_crew/slack/gateway.py @@ -206,7 +206,7 @@ ) from kiro_crew.mcp_hot_reload import parse_kiro_cli_version from kiro_crew.memory import MemoryStore -from kiro_crew.messaging import APPROVAL_INTERACTIVE, TurnDriver, registry +from kiro_crew.messaging import APPROVAL_INTERACTIVE, TurnDriver, inbound_spool, registry from kiro_crew.messaging.dispatch import build_directive_consumer, build_tool_gate from kiro_crew.messaging.display_safety import redact_for_display from kiro_crew.messaging.identity import channel_inbound_permitted, publish_turn_identity @@ -1814,6 +1814,9 @@ def __init__( # by messaging.registry.start_channels until the config-schema PR # retires them; shutdown closes through THIS dict. self._channel_handles: dict[str, object] = {} + # Detached boot task that replays the durable inbound spool (issue #2217). + # Held on the instance so the task is not garbage-collected mid-flight. + self._inbound_replay_task: "asyncio.Task[None] | None" = None self._model_download_task: "asyncio.Task[bool] | None" = None self._auto_migrate_task: "asyncio.Task[None] | None" = None # Boot-time update check, started fire-and-forget after the signal @@ -9505,6 +9508,17 @@ def _wire_mcp_gateway_dashboard(self) -> None: async def _shutdown(self) -> None: """Graceful cleanup of all services.""" + # Stop the boot-time inbound-spool notice pass before the transports it + # sends through are closed. Nothing is lost by cancelling: an entry is + # removed from disk only AFTER its notice is confirmed, so an entry cut + # off mid-send is noticed again on the next start (at most one duplicate + # line). Awaited with a small budget so a slow platform send cannot spend + # the GRACEFUL_SHUTDOWN_SECS that saves active chat slots. + replay = self._inbound_replay_task + if replay is not None and not replay.done(): + replay.cancel() + with contextlib.suppress(asyncio.CancelledError, asyncio.TimeoutError, Exception): + await asyncio.wait_for(replay, timeout=1.0) # Stop polling the central policy source, so a fetch in flight cannot # install a ceiling into a context the rest of this teardown is dismantling. # The join budget is deliberately small: the thread waits on an Event, so @@ -11747,6 +11761,31 @@ async def _start_channel_transports( # bailed out early. await loop.run_in_executor(maintenance_executor(), self._badge_unready_channels, boot) self._channel_handles = await registry.start_channels(self, descriptors, permitted) + # Tell the sender of whatever the SHUTDOWN GATE refused on the way down + # that it was never processed (issue #2217). Ordered AFTER the transports + # because the notice is sent through the channel that received the + # message, and detached from boot so a slow platform send cannot hold the + # gateway's start open. + self._inbound_replay_task = asyncio.create_task(self._replay_spooled_inbound()) + + async def _replay_spooled_inbound(self) -> None: + """Notice inbound messages the shutdown gate refused before this start. + + The spool is written only at the refusal point, so every entry is a turn + that provably never opened; the pass sends each sender an accurate + restart notice quoting their message and removes the entry only once the + send is confirmed — see :mod:`kiro_crew.messaging.inbound_spool`. + Entirely best-effort: this runs as a detached boot task, so an exception + escaping here would be an unretrieved task exception rather than + anything a user could act on. + """ + try: + transports = getattr(self.dashboard_state, "channel_transports", None) or {} + await inbound_spool.replay_spooled(transports=transports) + except asyncio.CancelledError: + raise + except Exception: + logger.warning("inbound spool: replay pass failed", exc_info=True) def _badge_unready_channels(self, bootable: "tuple[ChannelDescriptor, ...]") -> None: """Give an ENABLED channel that cannot start a reason the dashboard shows. diff --git a/src/kiro_crew/telegram/transport_dispatch.py b/src/kiro_crew/telegram/transport_dispatch.py index 0dd5eb63d94..63b1f20f8f2 100644 --- a/src/kiro_crew/telegram/transport_dispatch.py +++ b/src/kiro_crew/telegram/transport_dispatch.py @@ -62,6 +62,7 @@ ) from kiro_crew.messaging.driver import APPROVAL_INTERACTIVE, TurnDriver from kiro_crew.messaging.identity import channel_inbound_permitted, publish_turn_identity +from kiro_crew.messaging.inbound_spool import InboundRoute, spool_refused_turn from kiro_crew.messaging.link import ( CHAT_TYPE_DIRECT, CHAT_TYPE_FORUM, @@ -1108,6 +1109,37 @@ def _tool_gate(event: Any) -> str: "Telegram: aborting dispatch for %s — gateway is shutting down", session_key, ) + # Durable inbound spool (issue #2217). Written HERE and nowhere else: + # this is the one point where the payload is still in memory AND the + # turn is provably unopened, so a replay on the next start cannot + # double-answer a turn that actually ran. Telegram cannot recover this + # from its own offset either — ``_persistable_offset`` bounds duplicate + # replay, not loss, because the next long poll server-confirms the + # batch it just dispatched. + # + # NOT for a restricted session. ``/incognito`` and ``/temporary`` are a + # promise that this conversation persists nothing, and the spool is a + # durable file holding the message verbatim. The same predicate that + # gates the durable-history write gates this one; a refused restricted + # message degrades to the pre-feature loss, which is what the user asked + # for by choosing the mode. + if not await self._session_restricted(session_key): + await spool_refused_turn( + channel_type="telegram", + route=InboundRoute( + conversation_id=str(chat_id), + # ``msg.text``, NOT the local ``text``: by here the latter + # has attachment context appended, whose inlined temp paths + # are gone after a restart, and may have had a mid-turn + # override prefix stripped. The spool wants what the user + # typed. + text=msg.text, + user_id=str(user_id), + thread_id=str(reply_thread) if reply_thread else "", + message_id=str(getattr(msg, "message_id", "") or ""), + attachments_dropped=len(getattr(msg, "attachments", None) or ()), + ), + ) except Exception as exc: logger.exception("Telegram transport_dispatch: error handling message") # Permanent, user-actionable failures (e.g. model entitlement) diff --git a/src/kiro_crew/weixin/transport_dispatch.py b/src/kiro_crew/weixin/transport_dispatch.py index 28d7340dc73..0ba05c3cb07 100644 --- a/src/kiro_crew/weixin/transport_dispatch.py +++ b/src/kiro_crew/weixin/transport_dispatch.py @@ -45,6 +45,7 @@ inbound_permitted, ) from kiro_crew.messaging.driver import APPROVAL_INTERACTIVE +from kiro_crew.messaging.inbound_spool import InboundRoute from kiro_crew.messaging.link import build_dm_session_key, seed_generation from kiro_crew.messaging.transport import InboundMessage from kiro_crew.safety_override import safety_override @@ -182,6 +183,14 @@ async def handle_message(self, inbound: InboundMessage) -> None: # sender is told to resend once the turn ends, and any accompanying text # still reaches the turn via steer. attachment_temp_paths: list[str] = [] + # Captured BEFORE ingestion, which clears ``inbound.attachments`` and + # inlines the temp paths into the text. The durable inbound spool needs + # both originals (issue #2217): the count is what tells the restart + # notice this turn carried media that was not carried over, and the + # pre-ingestion text is what the notice quotes -- the ingested form + # holds paths to files that no longer exist. + original_text = text + original_attachments = len(inbound.attachments or ()) if inbound.attachments: ingested, attachment_temp_paths = await self._ingest_or_refuse(inbound, user_id, text) if ingested is None: @@ -190,7 +199,13 @@ async def handle_message(self, inbound: InboundMessage) -> None: inbound.text = text try: - await self._drive(inbound, user_id, text) + await self._drive( + inbound, + user_id, + text, + original_text=original_text, + original_attachments=original_attachments, + ) finally: if attachment_temp_paths: await asyncio.to_thread(cleanup_attachments, attachment_temp_paths) @@ -245,8 +260,22 @@ async def _say_resend_after_turn(self, user_id: str) -> None: """Tell a mid-turn sender their attachment needs resending.""" await self._say(user_id, _RESEND_AFTER_TURN) - async def _drive(self, inbound: InboundMessage, user_id: str, text: str) -> None: - """Session acquisition + turn dispatch for one already-ingested message.""" + async def _drive( + self, + inbound: InboundMessage, + user_id: str, + text: str, + *, + original_text: str = "", + original_attachments: int = 0, + ) -> None: + """Session acquisition + turn dispatch for one already-ingested message. + + ``original_text`` / ``original_attachments`` are the pre-ingestion values, + which this frame can no longer recover: ingestion clears + ``inbound.attachments`` and rewrites the text with temp paths that are gone + after a restart. They exist for the durable inbound spool (issue #2217). + """ assert self.client is not None # ── Mid-turn concurrency: check the CURRENT-generation key for an # in-flight turn BEFORE any idle/daily rotation (rotating first could @@ -283,6 +312,18 @@ async def _drive(self, inbound: InboundMessage, user_id: str, text: str) -> None ChannelTurn( channel_type="weixin", session_key=session_key, + # Durable inbound spool (issue #2217): the peer id IS the reply + # target on this DM-only channel, and the reply's context_token + # is already persisted off-loop, so the restart notice can land. + # ``original_text`` with NO fallback to the ingested ``text``: the + # ingested form inlines temp paths, and quoting those in the + # notice would disclose the crew's on-disk layout for nothing. + inbound_route=InboundRoute( + conversation_id=inbound.conversation_id, + text=original_text, + user_id=user_id, + attachments_dropped=original_attachments, + ), # Session-directive consumer: monitor_start / autonudge_stop / # ... return a marker TurnDriver decodes; apply it against THIS # turn's session key (dashboard-only directives stay refused diff --git a/src/kiro_crew/whatsapp/transport.py b/src/kiro_crew/whatsapp/transport.py index 360ecf691b5..8dc2d62c73b 100644 --- a/src/kiro_crew/whatsapp/transport.py +++ b/src/kiro_crew/whatsapp/transport.py @@ -174,6 +174,13 @@ def __init__( #: a channel-neutral shape and a WhatsApp stanza id means nothing to the #: others, so it rides here rather than widening the shared contract. self.pending_message_id: dict[int, str] = {} + #: ``(text as the user sent it, media count)`` captured BEFORE ingestion + #: rewrites ``msg.text`` with attachment context and temp paths. The + #: durable inbound spool (issue #2217) quotes the spooled text back to the + #: user in a restart notice, and the ingested form would quote on-disk + #: paths to files that no longer exist. Same keying and lifetime as the + #: three tables above. + self.pending_original: dict[int, tuple[str, int]] = {} client.on_message = self.receive # -- Tier-1 core --------------------------------------------------- @@ -498,6 +505,8 @@ async def receive(self, raw_envelope: Any) -> None: # the number trigger an authenticated download on the operator's host, # which is a remote-triggered fetch with no authorization behind it. temp_paths: list[str] = [] + # The user's own text and media count, before either is rewritten below. + self.pending_original[id(msg)] = (msg.text, 1 if desc.has_media else 0) if desc.has_media and not self._may_fetch_media(msg, is_group=is_group): # Refusing is said out loud: silence reads as the agent ignoring a # photo the sender believes it received. @@ -527,6 +536,7 @@ async def receive(self, raw_envelope: Any) -> None: self.pending_verdicts.pop(id(msg), None) self.pending_operator.pop(id(msg), None) self.pending_message_id.pop(id(msg), None) + self.pending_original.pop(id(msg), None) def _is_addressed(self, message: Any, chat: str, sender_is_operator: bool) -> bool: """Mentioned (@-tag of the linked account) or replying to the agent's diff --git a/src/kiro_crew/whatsapp/transport_dispatch.py b/src/kiro_crew/whatsapp/transport_dispatch.py index e493e01564c..be9d4c91434 100644 --- a/src/kiro_crew/whatsapp/transport_dispatch.py +++ b/src/kiro_crew/whatsapp/transport_dispatch.py @@ -30,6 +30,7 @@ inbound_permitted, ) from kiro_crew.messaging.driver import APPROVAL_INTERACTIVE +from kiro_crew.messaging.inbound_spool import InboundRoute from kiro_crew.messaging.link import build_dm_session_key, seed_generation from kiro_crew.messaging.transport import InboundMessage from kiro_crew.whatsapp.commands import ( @@ -264,6 +265,28 @@ async def _handle_compact(self, scope: str) -> None: finally: self.sessions.release(session_key) + def _inbound_route(self, inbound: InboundMessage) -> InboundRoute | None: + """The spool route for a DM: the user's own text and media count, or ``None``. + + Read from the transport's ``pending_original`` side table, which + ``receive`` fills BEFORE ingestion rewrites ``inbound.text``. Same lifetime + as ``pending_verdicts``. Deliberately NO fallback to ``inbound.text`` when + the entry is absent: a fallback to the ingested prompt is the disclosure + this table exists to prevent, so an envelope that did not come through + ``receive`` is simply not spooled. + """ + assert self.transport is not None + original = self.transport.pending_original.get(id(inbound)) + if original is None: + return None + text, media = original + return InboundRoute( + conversation_id=inbound.conversation_id, + text=text, + user_id=inbound.user_id, + attachments_dropped=media, + ) + async def _drive(self, inbound: InboundMessage, verdict: Any) -> None: assert self.transport is not None and self.client is not None transport = self.transport @@ -335,6 +358,23 @@ async def _drive(self, inbound: InboundMessage, verdict: Any) -> None: channel_type="whatsapp", session_key=session_key, conversation_id=f"whatsapp:{scope}", + # Durable inbound spool (issue #2217): DMs ONLY. The replay is a + # restart notice gated on ``may_send_to``, and this transport's + # ``may_send_to`` answers from ``dm_policy`` alone -- it knows + # nothing of the group roster, so a group removed or set to ``off`` + # while the gateway was down would still receive the notice. + # Rather than teach the egress gate a roster it was never asked to + # hold, group routes are not declared and a refused group message + # degrades exactly as before this seam (tracked in #9144). + # + # The text is the PRE-INGESTION original the transport captured, + # not ``inbound.text`` (by now rewritten with attachment context + # and temp paths that are dead after a restart) and not + # ``user_text`` (which can carry ``build_silence_contract`` -- + # private operating rules -- prepended to the model prompt). The + # notice quotes the spooled text back into the conversation, so + # only what the user actually sent may be spooled. + inbound_route=(None if group else self._inbound_route(inbound)), agent=agent, user_text=user_text, renderer=renderer, diff --git a/test/test_discord.py b/test/test_discord.py index 0046cce9165..946a29a3153 100644 --- a/test/test_discord.py +++ b/test/test_discord.py @@ -2271,6 +2271,35 @@ async def test_a_shutdown_between_the_claim_and_the_dispatch_never_opens_the_tur # Refused is not leaked -- the session-keyed semaphore still comes back. assert sess.released + @pytest.mark.asyncio + async def test_a_shutdown_refusal_is_not_spooled_for_a_restricted_session( + self, tmp_path, monkeypatch + ) -> None: + """An incognito or temporary conversation persists nothing, the spool included. + + RED-BEFORE: without the restricted-session gate at the refusal point the + private message is written verbatim to ``refused.jsonl``. + """ + from kiro_crew.messaging import inbound_spool as S + + monkeypatch.setattr(S, "data_home", lambda: tmp_path) + d, _cli, sess = _dispatcher({"u1"}) + sess.closing = True + spool = tmp_path / "inbound-spool" / "refused.jsonl" + + # Persistent: the refusal is spooled. + await d.handle_message(self._msg("keep me")) + assert spool.exists() and "keep me" in spool.read_text(encoding="utf-8") + spool.unlink() + + async def _restricted(_key: str) -> bool: + return True + + monkeypatch.setattr(d, "_session_restricted", _restricted) + await d.handle_message(self._msg("my secret")) + + assert not spool.exists(), "an incognito message was persisted to the spool" + @pytest.mark.asyncio async def test_monitor_wake_busy_at_dispatch_boundary_is_not_steered_or_queued( self, diff --git a/test/test_inbound_spool.py b/test/test_inbound_spool.py new file mode 100644 index 00000000000..6c3c10b7d20 --- /dev/null +++ b/test/test_inbound_spool.py @@ -0,0 +1,1444 @@ +"""Durable inbound spool: the loss it closes, and the bounds that keep it safe. + +Issue #2217. Gateway shutdown gathers channel teardown and +``SessionManager.close_all()`` concurrently, so a message the platform has +already accepted can be refused by the ``_closing`` gate before its turn ever +opens. Nothing retries it: the payload was discarded and the user was answered +with the channel's generic fault notice. + +The fix is deliberately narrow: record the refused message at the refusal +point, and on the next start tell the user, in that same conversation, that it +was never processed -- quoting it so a resend is one tap. It is NOT re-driven as +a turn; see the module docstring for why re-dispatch was built and removed. + +Two tests here are the RED-BEFORE pair, and both were proven to fail with the +production change reverted and the tests untouched: + +* :func:`test_a_refused_turn_is_spooled_with_its_routing` -- with the spool call + removed from ``drive_turn``'s ``except SessionClosingError`` branch it fails on + ``a refused message left no durable trace``, which is the loss itself. +* :func:`test_a_confirmed_notice_removes_the_entry` -- with the notice pass + unwired the spooled message is never answered and the entry never leaves disk. + +The rest pin the properties that make the feature safe rather than merely +present: opt-in per channel, a count cap, an age horizon, per-entry truncation, +dedupe of a double-spool, the egress re-authorization, at-least-once removal +(only after a CONFIRMED send), and the link/lock/fence rules that make the spool +a trust boundary rather than a writable input. +""" + +from __future__ import annotations + +import asyncio +import json +import os +from pathlib import Path +from typing import Any + +import pytest + +from kiro_crew.messaging import dispatch as D +from kiro_crew.messaging import inbound_spool as S +from kiro_crew.messaging.dispatch import ChannelTurn, drive_turn +from kiro_crew.messaging.inbound_spool import ( + SPOOL_MAX_ENTRIES, + InboundRoute, + SpooledInbound, + record_refusal_sync, + remove_entry, + replay_spooled, + spool_path, +) +from kiro_crew.messaging.transport import TransportCapabilities +from kiro_crew.session_allocation import SessionClosingError + +# ── Pipeline stand-ins, mirroring test_messaging_dispatch.py ────────────────── + + +class _Sessions: + """Refuses the turn at the closing gate, exactly as the real manager does.""" + + def __init__(self, closing: bool = True) -> None: + self.closing = closing + self.released = 0 + self.successes = 0 + self.failures = 0 + + async def get_or_create(self, key, agent=None, channel_id=None): + return object(), False, False + + def begin_turn(self, key): + if self.closing: + raise SessionClosingError("SessionManager is closing") + + async def set_channel(self, key, channel_id): + pass + + def record_success(self, key): + self.successes += 1 + + async def record_failure(self, key): + self.failures += 1 + + def release(self, key): + self.released += 1 + + def get_provider(self, key): + return object() + + +class _Renderer: + async def on_turn_start(self): + pass + + async def close(self): + pass + + +class _CtxBuilder: + def build_message(self, text, is_new, session_key, **kw): + return text, None + + +class _Driver: + last_stop_reason = "" + + def __init__(self, *a, **kw): + self._closing_gate = kw.get("closing_gate") + + async def run(self, message): + if self._closing_gate is not None: + self._closing_gate() + return "the reply" + + +def _patch_pipeline(monkeypatch) -> None: + async def _permitted(_channel_type): + return True + + async def _publish(_sessions, _key): + pass + + async def _embed(fn, *args, **kw): + return fn(*args, **kw) + + monkeypatch.setattr(D, "inbound_permitted", _permitted) + monkeypatch.setattr(D, "publish_turn_identity", _publish) + monkeypatch.setattr(D, "run_in_embed_pool", _embed) + monkeypatch.setattr(D, "TurnDriver", _Driver) + + +@pytest.fixture() +def spool_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path: + """Point the DEFAULT spool location at a temp home. + + ``spool_path`` resolves ``data_home`` per call (deliberately, so a HOME + override set after import is honoured), so patching the module attribute is + what redirects the paths the production call sites use — they take no + ``path`` argument. + """ + monkeypatch.setattr(S, "data_home", lambda: tmp_path) + return tmp_path / "inbound-spool" / "refused.jsonl" + + +def _turn(route: InboundRoute | None) -> ChannelTurn: + return ChannelTurn( + channel_type="weixin", + session_key="weixin:agentA:direct:userA", + conversation_id="weixin:userA", + agent="agentA", + user_text="what is the deploy status?", + renderer=_Renderer(), + approval_mode="auto", + inbound_route=route, + ) + + +def _entries(path: Path) -> list[dict[str, Any]]: + return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line] + + +def _texts(path: Path) -> list[str]: + return [row["text"] for row in _entries(path)] if path.exists() else [] + + +def peek_next(**kw: Any) -> SpooledInbound | None: + """The entry half of :func:`S.peek_next`; the held list is asserted where it matters.""" + entry, _held = S.peek_next(**kw) + return entry + + +class _Transport: + """A connected transport: an egress gate and a send that returns an id.""" + + capabilities = TransportCapabilities() + + def __init__( + self, + *, + may_send: bool = True, + send_raises: bool = False, + message_id: Any = "mid", + ) -> None: + self.sent: list[tuple[str, str, str | None]] = [] + self.send_gate_calls: list[tuple[str, str | None, str]] = [] + self._may_send = may_send + self._send_raises = send_raises + self._message_id = message_id + + def may_send_to(self, conversation_id, thread_id=None, *, principal=""): + self.send_gate_calls.append((conversation_id, thread_id, principal)) + return self._may_send + + async def send_message(self, conversation_id, content, thread_id=None): + if self._send_raises: + raise RuntimeError("conversation is gone") + self.sent.append((conversation_id, content, thread_id)) + return self._message_id + + +def _spool(path: Path, **kw: Any) -> SpooledInbound: + entry = SpooledInbound( + channel_type=kw.pop("channel_type", "telegram"), + conversation_id=kw.pop("conversation_id", "555"), + text=kw.pop("text", "please check CI"), + **kw, + ) + assert record_refusal_sync(entry, path=path) + return entry + + +# ── The loss (red-before on an unwired tree) ────────────────────────────────── + + +def test_a_refused_turn_is_spooled_with_its_routing(monkeypatch, spool_home: Path) -> None: + """The message survives a shutdown refusal instead of being discarded. + + RED-BEFORE: with no spool wired into ``drive_turn``'s + ``except SessionClosingError`` branch, the refusal leaves nothing on disk and + the user's text is gone for good — which is the whole of issue #2217. + """ + _patch_pipeline(monkeypatch) + sessions = _Sessions(closing=True) + + asyncio.run( + drive_turn( + _turn( + InboundRoute( + conversation_id="userA", user_id="userA", text="what is the deploy status?" + ) + ), + sessions=sessions, + ctx_builder=_CtxBuilder(), + ) + ) + + assert spool_home.exists(), "a refused message left no durable trace" + rows = _entries(spool_home) + assert len(rows) == 1 + assert rows[0]["text"] == "what is the deploy status?" + assert rows[0]["channel_type"] == "weixin" + assert rows[0]["conversation_id"] == "userA", ( + "the reply target must be the transport-addressable id, not the " "session-attribution id" + ) + assert rows[0]["user_id"] == "userA" + assert rows[0]["spooled_at"] > 0 + assert sessions.failures == 0, "a restart is not a session fault" + assert sessions.successes == 0, "no turn ran" + + +def test_a_channel_that_declares_no_route_is_not_spooled(monkeypatch, spool_home: Path) -> None: + """Adoption is opt-in, so an un-adopted channel is byte-identical to before. + + Without this, every channel would be half-enrolled: an entry written with no + addressable reply target posts the notice nowhere and looks like a silent + drop with extra disk writes. + """ + _patch_pipeline(monkeypatch) + + asyncio.run(drive_turn(_turn(None), sessions=_Sessions(), ctx_builder=_CtxBuilder())) + + assert not spool_home.exists() + + +def test_a_successful_turn_writes_nothing(monkeypatch, spool_home: Path) -> None: + """Zero cost on the happy path — the property that makes this affordable. + + It is also what makes the notice truthful: a turn that completed just before + exit was never written, so nobody is told to resend something that was + answered. + """ + _patch_pipeline(monkeypatch) + sessions = _Sessions(closing=False) + + asyncio.run( + drive_turn( + _turn(InboundRoute(conversation_id="userA")), + sessions=sessions, + ctx_builder=_CtxBuilder(), + ) + ) + + assert sessions.successes == 1 + assert not spool_home.exists() + + +def test_a_media_only_refusal_is_spooled_for_the_notice(spool_home: Path) -> None: + """An uncaptioned photo is still a message the platform marked delivered. + + Refusing to WRITE it made media-only the one message shape this spool + silently dropped. Recorded with an empty body and a nonzero dropped count, + which is what the notice names. + """ + wrote = asyncio.run( + S.spool_refused_turn( + channel_type="telegram", + route=InboundRoute(conversation_id="1", text=" ", attachments_dropped=2), + ) + ) + assert wrote is True + rows = _entries(spool_home) + assert rows[0]["attachments_dropped"] == 2 + + +def test_a_refusal_with_neither_text_nor_media_is_not_spooled(spool_home: Path) -> None: + """Nothing at all means nothing to notify about.""" + wrote = asyncio.run( + S.spool_refused_turn( + channel_type="telegram", + route=InboundRoute(conversation_id="1", text=" "), + ) + ) + assert wrote is False + assert not spool_home.exists() + + +# ── The notice pass ─────────────────────────────────────────────────────────── + + +def test_a_confirmed_notice_removes_the_entry(spool_home: Path) -> None: + """The fix, end to end: the user is told, in their conversation, to resend. + + RED-BEFORE: with the pass unwired the message is never answered and the + entry never leaves disk. + """ + _spool(spool_home, conversation_id="555", thread_id="7", text="please check CI") + transport = _Transport() + + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert report.notified and not report.dropped and not report.unconfirmed + assert len(transport.sent) == 1 + conversation, content, thread = transport.sent[0] + assert (conversation, thread) == ("555", "7"), "the notice went to the wrong conversation" + assert "> please check CI" in content, "the notice must quote the message so a resend is a tap" + assert "restarting" in content, "the notice must name the real cause, not a generic fault" + assert not spool_home.exists(), "a confirmed notice left the entry to be noticed again" + + +def test_a_dropped_attachment_is_named_in_the_notice(spool_home: Path) -> None: + _spool(spool_home, text="see attached", attachments_dropped=2) + transport = _Transport() + + asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert "2 attachment(s)" in transport.sent[0][1] + + +def test_a_media_only_entry_is_noticed(spool_home: Path) -> None: + _spool(spool_home, text="", attachments_dropped=1) + transport = _Transport() + + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert report.notified and "1 attachment(s)" in transport.sent[0][1] + + +def test_the_notice_is_delivered_at_least_once(spool_home: Path) -> None: + """An UNCONFIRMED send keeps the entry; the next start notices it again. + + This is the direction the notice-only design can afford: a duplicate notice + is a repeated line, never a repeated side effect. A raise and an empty + message id are both "unconfirmed" -- neither is evidence the user saw it. + """ + _spool(spool_home, message_id="a", text="raises") + _spool(spool_home, message_id="b", text="empty id") + + class _RaisesForOne(_Transport): + async def send_message(self, conversation_id, content, thread_id=None): + if "raises" in content: + raise RuntimeError("platform 5xx") + self.sent.append((conversation_id, content, thread_id)) + return "" + + transport = _RaisesForOne() + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert sorted(report.unconfirmed) == sorted([f"telegram:555:{i}" for i in ("a", "b")]) + assert not report.notified and not report.dropped + assert _texts(spool_home) == ["raises", "empty id"], "an unconfirmed notice removed the entry" + + # Next start, the platform is back: both are noticed and removed. + second = _Transport() + again = asyncio.run(replay_spooled(transports={"telegram": second})) + assert len(again.notified) == 2 and len(second.sent) == 2 + assert not spool_home.exists() + + +def test_an_unconfirmed_entry_does_not_block_the_rest(spool_home: Path) -> None: + """One bad route must not park the whole queue behind it.""" + _spool(spool_home, message_id="a", conversation_id="dead", text="dead route") + _spool(spool_home, message_id="b", conversation_id="live", text="live route") + + class _DeadRoute(_Transport): + async def send_message(self, conversation_id, content, thread_id=None): + if conversation_id == "dead": + raise RuntimeError("gone") + return await super().send_message(conversation_id, content, thread_id) + + transport = _DeadRoute() + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert report.unconfirmed == ["telegram:dead:a"] and report.notified == ["telegram:live:b"] + assert _texts(spool_home) == ["dead route"] + + +def test_a_transport_that_returns_no_ids_is_still_confirmed(spool_home: Path) -> None: + """WeCom and Feishu return ``""`` on SUCCESS and raise on failure. + + Reading their empty id as "unconfirmed" would notice the same message on + every start until the horizon. The capability is what says which idiom the + transport follows, so it is read rather than assumed. + """ + _spool(spool_home, channel_type="wecom") + + class _NoIds(_Transport): + capabilities = TransportCapabilities(returns_message_id=False) + + transport = _NoIds(message_id="") + report = asyncio.run(replay_spooled(transports={"wecom": transport})) + + assert report.notified and not report.unconfirmed + assert len(transport.sent) == 1 + assert not spool_home.exists() + + +def test_a_crash_between_send_and_removal_re_notices(spool_home: Path) -> None: + """The at-least-once half, driven through the real pass. + + The entry is removed only after the send returns, so a process that dies in + between leaves it on disk and the next start notices it again -- the user + sees a repeated line, and never a lost message. + """ + _spool(spool_home, message_id="m", text="once more") + + class _DiesAfterSend(_Transport): + async def send_message(self, conversation_id, content, thread_id=None): + self.sent.append((conversation_id, content, thread_id)) + raise KeyboardInterrupt("the gateway went down after the send") + + with pytest.raises(KeyboardInterrupt): + asyncio.run(replay_spooled(transports={"telegram": _DiesAfterSend()})) + assert _texts(spool_home) == ["once more"], "a crash after the send lost the entry" + + transport = _Transport() + asyncio.run(replay_spooled(transports={"telegram": transport})) + assert len(transport.sent) == 1 and not spool_home.exists() + + +def test_a_crash_on_the_first_entry_keeps_the_rest(spool_home: Path) -> None: + for index in range(3): + _spool(spool_home, message_id=f"m-{index}", text=f"message {index}") + + class _Dies(_Transport): + async def send_message(self, conversation_id, content, thread_id=None): + raise KeyboardInterrupt("mid-pass") + + with pytest.raises(KeyboardInterrupt): + asyncio.run(replay_spooled(transports={"telegram": _Dies()})) + + assert _texts(spool_home) == ["message 0", "message 1", "message 2"] + + +def test_a_second_pass_has_nothing_left_to_do(spool_home: Path) -> None: + _spool(spool_home) + transport = _Transport() + + asyncio.run(replay_spooled(transports={"telegram": transport})) + again = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert len(transport.sent) == 1 and again.total == 0 + + +def test_entries_are_noticed_oldest_first(spool_home: Path) -> None: + for index in range(3): + _spool(spool_home, message_id=f"m-{index}", text=f"message {index}") + transport = _Transport() + + asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert [content.splitlines()[-1] for _, content, _ in transport.sent] == [ + "> message 0", + "> message 1", + "> message 2", + ] + + +def test_an_over_cap_quote_is_truncated_visibly_not_sliced_by_the_transport( + spool_home: Path, +) -> None: + """The notice prefixes the quote, so a message that fit on the way in may not fit now. + + RED-BEFORE: with no sizing, a maximum-length message plus the prefix exceeds + ``max_message_chars``; a transport that slices to its cap and still returns an + id would confirm a notice whose tail was silently cut, and the entry would be + removed with text the user never saw echoed. Truncated HERE, and marked. + """ + _spool(spool_home, text="x" * 1900) + + class _Capped(_Transport): + capabilities = TransportCapabilities(max_message_chars=1900) + + transport = _Capped() + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert report.notified + content = transport.sent[0][1] + assert len(content) <= 1900, "the notice exceeded the transport cap" + assert "too long to quote in full" in content, "the truncation must be visible" + assert "restarting" in content, "the prefix must survive the cut" + + +def test_a_short_quote_is_not_truncated(spool_home: Path) -> None: + _spool(spool_home, text="short") + + class _Capped(_Transport): + capabilities = TransportCapabilities(max_message_chars=1900) + + transport = _Capped() + asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert "too long to quote" not in transport.sent[0][1] + assert "> short" in transport.sent[0][1] + + +def test_the_notice_is_made_display_safe_for_the_transport(spool_home: Path) -> None: + """A message can carry a broadcast mention; echoing it verbatim would fire it.""" + _spool(spool_home, channel_type="slack", text="@channel is CI down") + + transport = _Transport() + asyncio.run(replay_spooled(transports={"slack": transport})) + + content = transport.sent[0][1] + assert "@channel is CI down" not in content, "a broadcast mention was echoed live" + assert "@\u200bchannel is CI down" in content + + +# ── Holding: an absent channel is not a revoked one ─────────────────────────── + + +def test_an_unconnected_channel_entry_is_held_for_the_next_start(spool_home: Path) -> None: + """A channel absent THIS run may have merely failed to start. + + Dropping the entry turned a transient startup failure -- a token refresh that + timed out, a socket that came up late -- into permanent loss of a message the + platform marked delivered. Holding costs nothing, because the age horizon + still expires an entry for a channel the operator genuinely turned off. + """ + _spool(spool_home, channel_type="discord", text="still here?") + + report = asyncio.run(replay_spooled(transports={})) + + assert report.held and not report.dropped and not report.notified + assert _texts(spool_home) == [ + "still here?" + ], "the entry was discarded on a run where its channel was simply not up" + + transport = _Transport() + second = asyncio.run(replay_spooled(transports={"discord": transport})) + assert second.notified and len(transport.sent) == 1 + + +def test_a_held_entry_never_leaves_disk_during_the_pass(spool_home: Path) -> None: + """The predicate runs UNDER the lock and skips in place; nothing is requeued.""" + _spool(spool_home, channel_type="discord", message_id="d", text="held") + _spool(spool_home, channel_type="telegram", message_id="t", text="live") + seen_on_disk: list[list[str]] = [] + + class _Observing(_Transport): + async def send_message(self, conversation_id, content, thread_id=None): + seen_on_disk.append(_texts(spool_home)) + return await super().send_message(conversation_id, content, thread_id) + + asyncio.run(replay_spooled(transports={"telegram": _Observing()})) + + assert seen_on_disk == [["held", "live"]], "an entry left disk before its send was confirmed" + assert _texts(spool_home) == ["held"] + + +def test_a_held_entry_keeps_its_original_arrival_order(spool_home: Path) -> None: + for index in range(3): + _spool(spool_home, channel_type="discord", message_id=f"m-{index}", text=f"m {index}") + + asyncio.run(replay_spooled(transports={})) + + assert _texts(spool_home) == ["m 0", "m 1", "m 2"] + + +def test_a_held_entry_still_expires_on_the_age_horizon(spool_home: Path) -> None: + """Holding must not become an unbounded queue: the horizon is the bound.""" + assert record_refusal_sync( + SpooledInbound(channel_type="discord", conversation_id="1", text="old"), + path=spool_home, + now=1_000.0, + ) + asyncio.run(replay_spooled(transports={}, now=1_000.0 + 10)) + assert spool_home.exists() + + asyncio.run(replay_spooled(transports={}, now=1_000.0 + S.SPOOL_MAX_AGE_SECS + 1)) + assert not spool_home.exists(), "a held entry outlived the horizon" + + +# ── Authorization is re-decided at replay time ─────────────────────────────── + + +def test_the_notice_is_withheld_from_a_revoked_conversation(spool_home: Path) -> None: + """A spooled entry is not a standing grant for the conversation it names. + + The gateway was down for the whole window this feature spans, so the peer may + have left the roster or the Topic been de-allow-listed since. The notice is a + proactive send, which is exactly what ``may_send_to`` governs -- and a revoked + entry is REMOVED, not held: there is nothing this module may ever do with it. + """ + _spool(spool_home, channel_type="teams", conversation_id="conv-1", user_id="u-1") + transport = _Transport(may_send=False) + + report = asyncio.run(replay_spooled(transports={"teams": transport})) + + assert transport.sent == [], "output reached a conversation that is no longer authorized" + assert report.dropped and not report.notified + assert transport.send_gate_calls == [("conv-1", None, "u-1")], ( + "the principal must be passed, or a transport that authorizes by peer " + "cannot make the decision" + ) + assert not spool_home.exists() + + +def test_a_revoked_route_drop_is_audited(spool_home: Path, monkeypatch) -> None: + """The ``may_send_to`` denial is a permission decision that also deletes a message. + + RED-BEFORE: without the audit the SEL trail shows the channel scope granted + and then nothing, so a revoked route silently losing its notice looks exactly + like nothing having been spooled. Same record shape as the cross-surface + proactive send's identical denial. + """ + recorded: list[dict[str, Any]] = [] + monkeypatch.setattr( + S, + "sel", + lambda: type("Sel", (), {"log_api_access": lambda self, **kw: recorded.append(kw)})(), + ) + _spool(spool_home, channel_type="teams", conversation_id="conv-1", user_id="u-1") + + asyncio.run(replay_spooled(transports={"teams": _Transport(may_send=False)})) + + assert len(recorded) == 1, "the denial was not audited" + assert recorded[0]["operation"] == "channel.proactive_send_authorize" + assert recorded[0]["outcome"] == "denied" + assert recorded[0]["source"] == "teams" + assert recorded[0]["caller"] == "u-1" + + +def test_an_allowed_route_is_audited_too(spool_home: Path, monkeypatch) -> None: + """The grant is the decision that puts the gateway's own text into a conversation.""" + recorded: list[dict[str, Any]] = [] + monkeypatch.setattr( + S, + "sel", + lambda: type("Sel", (), {"log_api_access": lambda self, **kw: recorded.append(kw)})(), + ) + _spool(spool_home, channel_type="teams", conversation_id="conv-1", user_id="u-1") + + asyncio.run(replay_spooled(transports={"teams": _Transport()})) + + outcomes = [(r["operation"], r["outcome"]) for r in recorded] + assert ("channel.proactive_send_authorize", "allowed") in outcomes + + +def test_a_transport_with_no_egress_gate_is_refused_the_notice(spool_home: Path) -> None: + """Fails closed: a spool entry is the one input that did not come from the platform.""" + + class _NoGate: + capabilities = TransportCapabilities() + + def __init__(self) -> None: + self.sent: list[Any] = [] + + async def send_message(self, conversation_id, content, thread_id=None): + self.sent.append(conversation_id) + return "mid" + + _spool(spool_home, channel_type="teams") + transport = _NoGate() + + report = asyncio.run(replay_spooled(transports={"teams": transport})) + + assert transport.sent == [] + assert report.dropped + + +def test_a_channel_denied_by_governance_gets_no_notice_and_is_held( + spool_home: Path, monkeypatch +) -> None: + """The ``channels`` policy ceiling is asked before every notice, fail-closed. + + RED-BEFORE: without the vet, a channel the operator denied while the gateway + was down still receives the notice, because the transport is CONNECTED and + ``may_send_to`` answers only about the route. Held rather than dropped: the + route is not revoked, the channel is governed off, and the horizon bounds it. + """ + calls: list[tuple[str, str, str]] = [] + + def deny(scope, item, *, session_key, tool_name, **kw): + calls.append((scope, item, tool_name)) + return type("D", (), {"permitted": False})() + + # Patched where it is BOUND (module-scope import), as the top-level-imports + # rule requires; patching the defining module would leave the bound name. + monkeypatch.setattr(S, "vet_and_audit", deny) + _spool(spool_home, channel_type="telegram") + transport = _Transport() + + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert transport.sent == [], "a governance-denied channel received the notice" + assert report.held and not report.notified and not report.dropped + assert calls == [("channels", "telegram", "inbound_spool.notice")] + assert _texts(spool_home) == ["please check CI"], "a held entry must stay on disk" + + +def test_a_governance_evaluation_failure_denies(spool_home: Path, monkeypatch) -> None: + def boom(*a, **kw): + raise RuntimeError("profile store unavailable") + + monkeypatch.setattr(S, "vet_and_audit", boom) + _spool(spool_home) + transport = _Transport() + + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert transport.sent == [] and report.held + + +def test_a_raising_egress_gate_is_read_as_revoked(spool_home: Path) -> None: + class _Raises(_Transport): + def may_send_to(self, conversation_id, thread_id=None, *, principal=""): + raise RuntimeError("roster unavailable") + + _spool(spool_home) + transport = _Raises() + + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert transport.sent == [] and report.dropped + + +def test_a_threaded_route_is_authorized_by_the_thread_roster_alone(spool_home: Path) -> None: + """The principal is withheld for a threaded route; a DM route still carries it. + + Discord's ``may_send_to`` falls from a thread not in ``_allowed_threads`` to + ``principal in _allowed``, assuming a thread route names no principal. A spooled + thread entry names the sender, so passing it would let a still-allowed sender + authorize a notice into a revoked thread. + """ + _spool(spool_home, channel_type="discord", conversation_id="thr", thread_id="thr", user_id="u") + _spool(spool_home, channel_type="discord", conversation_id="dm", user_id="u", message_id="m") + transport = _Transport() + + asyncio.run(replay_spooled(transports={"discord": transport})) + + assert sorted(transport.send_gate_calls) == sorted([("thr", "thr", ""), ("dm", None, "u")]) + + +def test_a_revoked_discord_thread_gets_no_notice_even_from_an_allowed_sender( + spool_home: Path, +) -> None: + """Against the REAL Discord egress gate, not a stand-in. + + RED-BEFORE: with ``principal=entry.user_id`` passed for the thread route, the + still-allowed sender authorizes via the DM arm and the notice lands in a + thread that is no longer on the roster. + """ + from kiro_crew.discord.transport import DiscordTransport + + class _Client: + pass + + transport = DiscordTransport(_Client(), allowed_user_ids=["42"]) # type: ignore[arg-type] + # No allowed threads: the spooled thread was auto-created, memory-only, and + # is gone after the restart -- the ordinary post-restart state. + sent: list[tuple[str, str | None]] = [] + + async def send_message(conversation_id, content, thread_id=None): + sent.append((conversation_id, thread_id)) + return "mid" + + transport.send_message = send_message # type: ignore[method-assign] + _spool( + spool_home, channel_type="discord", conversation_id="9001", thread_id="9001", user_id="42" + ) + + report = asyncio.run(replay_spooled(transports={"discord": transport})) + + assert sent == [], "the notice was posted into a thread the roster no longer allows" + assert report.dropped and not report.notified + + +# ── Dedupe: a double-spool of the SAME message, and nothing more ───────────── + + +def test_the_same_message_spooled_twice_yields_one_entry(spool_home: Path) -> None: + """A retry loop around the refusal point must not multiply the entry.""" + for _ in range(3): + _spool(spool_home, message_id="m-1") + + assert len(_entries(spool_home)) == 1 + + +def test_two_messages_with_no_platform_id_stay_distinct(spool_home: Path) -> None: + """An entry the platform gave no id is never collapsed against another.""" + _spool(spool_home, text="first") + _spool(spool_home, text="second") + + assert set(_texts(spool_home)) == {"first", "second"} + + +def test_two_identical_bodies_with_no_platform_id_both_survive(spool_home: Path) -> None: + """Repeating yourself is ordinary; losing the second message is not. + + A body digest is the obvious-looking identity for a channel with no message + id, and it is a data-loss bug: two identical messages hash the same, so one + accepted message would be silently discarded. The hazard it would guard + against does not exist -- the refusal is one ``except`` branch that runs once + per turn. + """ + _spool(spool_home, channel_type="weixin", conversation_id="peer-1", text="status") + _spool(spool_home, channel_type="weixin", conversation_id="peer-1", text="status") + + assert _texts(spool_home) == [ + "status", + "status", + ], "the second 'status' was discarded — the user asked twice and is told once" + + +def test_two_identical_id_less_bodies_are_each_noticed_once(spool_home: Path) -> None: + """Removal is by ONE occurrence, so a shared trace id removes one, not both.""" + _spool(spool_home, channel_type="weixin", conversation_id="peer-1", text="status") + _spool(spool_home, channel_type="weixin", conversation_id="peer-1", text="status") + transport = _Transport() + + report = asyncio.run(replay_spooled(transports={"weixin": transport})) + + assert len(transport.sent) == 2 and len(report.notified) == 2 + assert not spool_home.exists() + + +def test_an_entry_with_no_platform_id_has_no_dedupe_identity(spool_home: Path) -> None: + """The empty key is the contract, not an accident of formatting.""" + without = SpooledInbound(channel_type="weixin", conversation_id="p", text="hi") + with_id = SpooledInbound( + channel_type="telegram", conversation_id="1", text="hi", message_id="m-1" + ) + + assert without.dedupe_key == "" + assert with_id.dedupe_key == "telegram:1:m-1" + assert without.trace_id.startswith("weixin:p:") + + +# ── Bounds: a crash-loop must not grow the spool ───────────────────────────── + + +def test_the_count_cap_keeps_the_newest_entries(spool_home: Path) -> None: + """Newest-wins, so a wedged gateway cannot accumulate a notice storm.""" + for index in range(SPOOL_MAX_ENTRIES + 10): + _spool(spool_home, message_id=f"m-{index}", text=f"message {index}") + + rows = _entries(spool_home) + assert len(rows) == SPOOL_MAX_ENTRIES + assert rows[-1]["text"] == f"message {SPOOL_MAX_ENTRIES + 9}" + assert rows[0]["text"] == "message 10", "the oldest entries should be the ones dropped" + + +def test_the_age_horizon_drops_a_stale_entry_on_read(spool_home: Path) -> None: + """A day-old unanswered message is noise, and the horizon is what expires it. + + Applied on READ as well as on write, because an entry that was legal when + written can sit past the horizon while the gateway is down. + """ + assert record_refusal_sync( + SpooledInbound(channel_type="telegram", conversation_id="1", text="old"), + path=spool_home, + now=1_000.0, + ) + + assert peek_next(path=spool_home, now=1_000.0 + S.SPOOL_MAX_AGE_SECS + 1) is None + assert not spool_home.exists(), "an all-stale spool must not be re-read on every start" + + +def test_a_stale_entry_does_not_survive_a_later_write(spool_home: Path) -> None: + assert record_refusal_sync( + SpooledInbound(channel_type="telegram", conversation_id="1", text="old", message_id="a"), + path=spool_home, + now=1_000.0, + ) + assert record_refusal_sync( + SpooledInbound(channel_type="telegram", conversation_id="1", text="new", message_id="b"), + path=spool_home, + now=1_000.0 + S.SPOOL_MAX_AGE_SECS + 1, + ) + + assert _texts(spool_home) == ["new"] + + +def test_an_over_cap_body_is_truncated_and_says_so(spool_home: Path) -> None: + """One paste must not be the whole budget, and truncation must be visible.""" + _spool(spool_home, text="x" * (S.TEXT_CAP * 2)) + + text = _entries(spool_home)[0]["text"] + assert len(text) <= S.TEXT_CAP + assert "truncated" in text + + +def test_the_pass_is_bounded_by_the_count_cap(spool_home: Path) -> None: + """A transport that never confirms cannot spin the pass forever.""" + for index in range(5): + _spool(spool_home, message_id=f"m-{index}") + transport = _Transport(message_id="") + + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert len(transport.sent) == 5, "each unconfirmed entry is attempted exactly once per pass" + assert len(report.unconfirmed) == 5 + + +# ── The store primitives ────────────────────────────────────────────────────── + + +def test_peek_returns_the_oldest_without_removing_it(spool_home: Path) -> None: + for index in range(3): + _spool(spool_home, message_id=f"m-{index}", text=f"message {index}") + + first = peek_next(path=spool_home) + + assert first is not None and first.text == "message 0", "oldest first" + assert _texts(spool_home) == ["message 0", "message 1", "message 2"], "peek removed an entry" + + +def test_peek_holds_an_unconnected_channel_and_skips_a_seen_entry(spool_home: Path) -> None: + held_entry = _spool(spool_home, channel_type="discord", message_id="d", text="held") + seen_entry = _spool(spool_home, channel_type="telegram", message_id="s", text="seen") + _spool(spool_home, channel_type="telegram", message_id="t", text="take") + + entry, held = S.peek_next(path=spool_home, connected={"telegram"}, skip={seen_entry.trace_id}) + + assert entry is not None and entry.text == "take" + assert held == [held_entry.trace_id], "a held entry must be reported, not silently skipped" + assert _texts(spool_home) == ["held", "seen", "take"], "peek removed an entry" + + +def test_a_failed_removal_notices_once_per_pass_not_once_per_iteration( + spool_home: Path, monkeypatch +) -> None: + """A spool that cannot be rewritten must not turn one entry into 128 notices. + + RED-BEFORE: with the removal result ignored, the entry stays on disk and + actionable, ``peek_next`` returns it again, and the bounded loop sends the + same notice ``SPOOL_MAX_ENTRIES`` times in one pass. The pass now remembers + every entry it has acted on; the next start retries the removal. + """ + _spool(spool_home, message_id="a", text="once, please") + _spool(spool_home, message_id="b", text="me too") + monkeypatch.setattr(S, "remove_entry", lambda entry, *, path=None: False) + transport = _Transport() + + report = asyncio.run(replay_spooled(transports={"telegram": transport})) + + assert len(transport.sent) == 2, "an entry was noticed more than once in a single pass" + assert sorted(report.unconfirmed) == sorted(["telegram:555:a", "telegram:555:b"]) + assert not report.notified, "a notice whose entry is still on disk is not 'done'" + assert _texts(spool_home) == ["once, please", "me too"] + + +def test_remove_takes_exactly_one_entry(spool_home: Path) -> None: + entries = [ + _spool(spool_home, message_id=f"m-{index}", text=f"message {index}") for index in range(3) + ] + + assert remove_entry(entries[1], path=spool_home) is True + assert _texts(spool_home) == ["message 0", "message 2"] + assert remove_entry(entries[1], path=spool_home) is False, "removing twice must be a no-op" + + +def test_the_spool_file_is_removed_once_drained(spool_home: Path) -> None: + """An emptied spool must not sit on disk being re-read on every start.""" + entry = _spool(spool_home) + + assert remove_entry(entry, path=spool_home) is True + assert not spool_home.exists() + assert peek_next(path=spool_home) is None + + +def test_a_failed_unlink_of_the_last_entry_does_not_resurrect_it( + spool_home: Path, monkeypatch +) -> None: + """The removal is the atomic replace, not the unlink. + + On Windows an AV scanner or indexer holding a handle makes ``unlink`` fail + routinely. If the last entry were removed by unlink alone, that failure would + leave it on disk to be noticed again on every start until the horizon. So the + remainder is always written (empty when nothing is left) and the unlink is a + tidy-up that gates nothing. + """ + entry = _spool(spool_home, message_id="only", text="once") + real_unlink = Path.unlink + + def _sharing_violation(self, *args, **kwargs): + if self == spool_home: + raise PermissionError(32, "sharing violation") + return real_unlink(self, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", _sharing_violation) + + assert remove_entry(entry, path=spool_home) is True + assert peek_next(path=spool_home) is None, "the removed entry came back after a failed unlink" + + +# ── Robustness: this runs during shutdown and during boot ──────────────────── + + +def test_an_unreadable_spool_is_left_untouched(spool_home: Path, monkeypatch) -> None: + """A transient read failure must defer the pass, not erase the queue. + + Every writer rewrites the file from what it read, and the reader unlinks a + file it read as empty. Reporting EIO as ``[]`` therefore had the next read + delete every queued message. + """ + _spool(spool_home, text="keep me") + real_open = S._open_own_file + + def _flaky_open(path): + if Path(path) == spool_home: + raise PermissionError(13, "transient") + return real_open(path) + + monkeypatch.setattr(S, "_open_own_file", _flaky_open) + assert peek_next(path=spool_home) is None + monkeypatch.undo() + + assert spool_home.exists(), "the spool was deleted after a read failure" + assert _texts(spool_home) == ["keep me"] + + +def test_an_unreadable_spool_refuses_a_write_rather_than_overwriting( + spool_home: Path, monkeypatch +) -> None: + _spool(spool_home, text="keep me") + real_open = S._open_own_file + + def _flaky_open(path): + if Path(path) == spool_home: + raise PermissionError(13, "transient") + return real_open(path) + + monkeypatch.setattr(S, "_open_own_file", _flaky_open) + wrote = record_refusal_sync( + SpooledInbound(channel_type="telegram", conversation_id="1", text="new"), + path=spool_home, + ) + monkeypatch.undo() + + assert wrote is False + assert _texts(spool_home) == ["keep me"] + + +def test_a_write_failure_degrades_to_the_old_loss(tmp_path: Path) -> None: + """Never the thing that fails shutdown — a refusal to write is just today.""" + blocked = tmp_path / "not-a-dir" + blocked.write_text("i am a file", encoding="utf-8") + + wrote = record_refusal_sync( + SpooledInbound(channel_type="telegram", conversation_id="1", text="hi"), + path=blocked / "spool.jsonl", + ) + + assert wrote is False + + +def test_a_corrupt_line_is_skipped_rather_than_failing_the_boot(spool_home: Path) -> None: + """The pass runs on the boot path, so a hand-edited spool must not raise.""" + spool_home.parent.mkdir(parents=True, exist_ok=True) + good = json.dumps( + SpooledInbound( + channel_type="telegram", conversation_id="1", text="ok", spooled_at=9e9 + ).to_dict() + ) + spool_home.write_text(f"not json\n{good}\n{{}}\n", encoding="utf-8") + + entry = peek_next(path=spool_home, now=9e9) + + assert entry is not None and entry.text == "ok" + + +def test_an_entry_with_no_reply_target_is_refused(spool_home: Path) -> None: + """Fails closed: an unaddressable entry would send the notice nowhere.""" + assert SpooledInbound.from_dict({"channel_type": "telegram", "text": "hi"}) is None + assert SpooledInbound.from_dict({"conversation_id": "1", "text": "hi"}) is None + assert SpooledInbound.from_dict({"channel_type": "telegram", "conversation_id": "1"}) is None + + +def test_the_default_spool_path_lives_under_the_data_home(spool_home: Path) -> None: + assert spool_path() == spool_home + + +def test_an_entry_records_only_what_the_notice_reads() -> None: + """No field ridden along "for later": every persisted key is consumed by the pass. + + ``session_key`` and ``chat_type`` were recorded for a re-dispatch design that + was removed (#9144). A field nothing reads is a field nothing tests. + """ + keys = set(SpooledInbound(channel_type="t", conversation_id="c", text="x").to_dict()) + assert keys == { + "channel_type", + "conversation_id", + "text", + "user_id", + "thread_id", + "message_id", + "attachments_dropped", + "spooled_at", + } + # A record from the earlier shape still parses: unknown keys are ignored. + legacy = {"channel_type": "t", "conversation_id": "c", "text": "x", "session_key": "s"} + assert SpooledInbound.from_dict(legacy) is not None + + +def test_the_pass_never_raises_on_a_transport_fault(spool_home: Path) -> None: + """A send that raises is "unconfirmed", not a boot failure.""" + _spool(spool_home) + + report = asyncio.run(replay_spooled(transports={"telegram": _Transport(send_raises=True)})) + + assert report.unconfirmed and _texts(spool_home) == ["please check CI"] + + +def test_a_cancelled_refusal_handler_still_lands_the_write(spool_home: Path) -> None: + """The write is off-loop AND survives the caller's cancellation. + + The handler that reaches the refusal is a task ``close_all`` is about to + cancel. A bare ``await asyncio.to_thread(...)`` there is a cancellation point + that would orphan the write; blocking the loop instead would stall a shutdown + already racing a deadline. ``asyncio.shield`` keeps both: the caller is + cancelled, the write is not, and the entry is on disk when the loop drains. + """ + import threading + + started = threading.Event() + release = threading.Event() + real_sync = S.record_refusal_sync + + def slow_sync(entry, *, path=None, now=None): + started.set() + assert release.wait(5), "test harness never released the write" + return real_sync(entry, path=path, now=now) + + async def scenario() -> None: + S.record_refusal_sync = slow_sync # type: ignore[assignment] + try: + task = asyncio.create_task( + S.record_refusal( + SpooledInbound(channel_type="telegram", conversation_id="1", text="survive") + ) + ) + await asyncio.to_thread(started.wait, 5) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert not spool_home.exists(), "the write had not been released yet" + release.set() + # Let the shielded inner task finish before the loop closes. + for _ in range(50): + if spool_home.exists(): + break + await asyncio.sleep(0.02) + finally: + S.record_refusal_sync = real_sync # type: ignore[assignment] + + asyncio.run(scenario()) + + assert _texts(spool_home) == ["survive"], "the caller's cancel orphaned the write" + + +def test_two_concurrent_refusals_both_survive(spool_home: Path) -> None: + """The read-modify-write is serialized, so neither message is overwritten.""" + import concurrent.futures + + entries = [ + SpooledInbound( + channel_type="telegram", + conversation_id="1", + text=f"message {index}", + message_id=f"m-{index}", + ) + for index in range(24) + ] + + with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool: + results = list(pool.map(lambda entry: record_refusal_sync(entry, path=spool_home), entries)) + + assert all(results) + assert set(_texts(spool_home)) == { + f"message {index}" for index in range(24) + }, "a concurrent refusal overwrote another message" + + +# ── The spool is a trust boundary ───────────────────────────────────────────── + + +def test_a_linked_spool_directory_is_refused_for_read_and_write(tmp_path: Path) -> None: + """A link planted before the fence existed must not be followed. + + The fence only holds from the build that ships it. A same-UID agent on an + older build could plant a symlink at ``inbound-spool``; every open in this + module would then resolve inside the link's target, which the fence never + covered, and a JSON record forged there would post a notice quoting text the + user never sent into a conversation of the forger's choosing. So a linked + directory is refused BEFORE the lock, the read, the write or the unlink. + """ + elsewhere = tmp_path / "elsewhere" + elsewhere.mkdir() + forged = SpooledInbound( + channel_type="telegram", conversation_id="1", text="forged", spooled_at=9e9 + ) + (elsewhere / "refused.jsonl").write_text(json.dumps(forged.to_dict()) + "\n", encoding="utf-8") + home = tmp_path / "home" + home.mkdir() + (home / "inbound-spool").symlink_to(elsewhere, target_is_directory=True) + spool = home / "inbound-spool" / "refused.jsonl" + + assert peek_next(path=spool, now=9e9) is None, "a forged record behind a link was read" + assert (elsewhere / "refused.jsonl").exists(), "the link's target was mutated" + assert ( + record_refusal_sync( + SpooledInbound(channel_type="telegram", conversation_id="1", text="hi"), path=spool + ) + is False + ) + assert (elsewhere / "refused.jsonl").read_text(encoding="utf-8").count( + "\n" + ) == 1, "a write through the linked directory landed in its target" + + +def test_a_linked_spool_leaf_is_refused(tmp_path: Path) -> None: + """The leaf too: ``os.open`` follows a final-component link.""" + elsewhere = tmp_path / "elsewhere.jsonl" + forged = SpooledInbound( + channel_type="telegram", conversation_id="1", text="forged", spooled_at=9e9 + ) + elsewhere.write_text(json.dumps(forged.to_dict()) + "\n", encoding="utf-8") + spool_dir = tmp_path / "inbound-spool" + spool_dir.mkdir() + spool = spool_dir / "refused.jsonl" + spool.symlink_to(elsewhere) + + assert peek_next(path=spool, now=9e9) is None + assert elsewhere.exists() + assert ( + record_refusal_sync( + SpooledInbound(channel_type="telegram", conversation_id="1", text="hi"), path=spool + ) + is False + ) + + +def test_a_hard_linked_spool_leaf_is_refused(tmp_path: Path) -> None: + """A second name for the inode is another way to feed this module bytes it did not write.""" + spool_dir = tmp_path / "inbound-spool" + spool_dir.mkdir() + spool = spool_dir / "refused.jsonl" + assert record_refusal_sync( + SpooledInbound(channel_type="telegram", conversation_id="1", text="hi"), path=spool + ) + os.link(spool, tmp_path / "alias.jsonl") + + assert peek_next(path=spool) is None + assert spool.exists(), "the hard-linked spool was mutated" + + +def test_every_spool_write_is_owner_only() -> None: + """Every rewrite goes through ``atomic_write(restrict_to_owner=True)``. + + Another same-UID reader is the threat model, and a rewrite that forgot the + flag would widen the mode on the very next pass. + """ + import inspect + + for fn in (S.record_refusal_sync, S.peek_next, S.remove_entry): + source = inspect.getsource(fn) + assert source.count("atomic_write(") == source.count("restrict_to_owner=True"), fn.__name__ + assert "restrict_to_owner=True" in source, fn.__name__ + + +def test_the_spool_directory_is_fenced_from_agent_file_tools() -> None: + """A file an agent could write is a way to post a notice as the gateway. + + Each entry names a conversation and carries text the notice quotes into it + verbatim. The egress recheck narrows the forge path to conversations still + authorized, which is not a boundary — the boundary is the file being + unreachable. Read matters too: an entry holds the verbatim text of a message + the operator sent. + """ + from kiro_crew.security import _CREW_SECRET_LEAVES + + assert "inbound-spool" in _CREW_SECRET_LEAVES + assert spool_path().parent.name == "inbound-spool", ( + "the fence entry is directory-scoped, so the spool must live in that " + "directory and not merely be named after it" + ) + + +def test_the_spool_directory_is_masked_in_agent_sandboxes() -> None: + """The file fence covers tools; this covers a spawned command.""" + from kiro_crew.sandbox import _CREW_HIDDEN_LEAVES + + assert "inbound-spool" in _CREW_HIDDEN_LEAVES + + +# ── Channel wiring: what each adopter declares as its route ────────────────── + + +def test_the_spooled_text_is_the_user_message_not_the_model_prompt( + monkeypatch, spool_home: Path +) -> None: + """A transformed prompt must never be what gets spooled or quoted back. + + WhatsApp's rules mode prepends the group's private operating rules and its + silence instructions to the model prompt. Spooling ``ChannelTurn.user_text`` + would put those rules in the entry, and the restart notice quotes the entry + verbatim — publishing them into the conversation. + """ + _patch_pipeline(monkeypatch) + turn = ChannelTurn( + channel_type="whatsapp", + session_key="whatsapp:agentA:direct:p1", + conversation_id="whatsapp:p1", + agent="agentA", + user_text="\n\nwhen is the deploy?", + renderer=_Renderer(), + approval_mode="auto", + inbound_route=InboundRoute(conversation_id="p1", text="when is the deploy?"), + ) + + asyncio.run(drive_turn(turn, sessions=_Sessions(), ctx_builder=_CtxBuilder())) + + text = _entries(spool_home)[0]["text"] + assert text == "when is the deploy?" + assert "private rules" not in text, "the operator's rules were spooled" + + +def test_the_turn_prompt_is_never_a_fallback_for_a_route_with_no_text( + monkeypatch, spool_home: Path +) -> None: + """An empty route text does NOT reach for ``ChannelTurn.user_text``. + + The fallback looked like a convenience for a channel whose two strings are + identical, and it was a disclosure: a media-only rules-mode message has an + EMPTY route text and a ``user_text`` that is the private rules -- so the + fallback spooled the rules and the notice quoted them. With no attachments and + no text there is nothing to spool; the prompt is never it. + """ + _patch_pipeline(monkeypatch) + turn = ChannelTurn( + channel_type="whatsapp", + session_key="whatsapp:agentA:direct:p1", + conversation_id="whatsapp:p1", + agent="agentA", + user_text="\n\n", + renderer=_Renderer(), + approval_mode="auto", + inbound_route=InboundRoute(conversation_id="p1"), + ) + + asyncio.run(drive_turn(turn, sessions=_Sessions(), ctx_builder=_CtxBuilder())) + + assert not spool_home.exists(), "the rules were spooled via the prompt fallback" + + +def test_a_media_only_rules_mode_message_spools_no_rules(monkeypatch, spool_home: Path) -> None: + """Uncaptioned photo, media denied, rules-mode prompt: only the count is kept.""" + _patch_pipeline(monkeypatch) + turn = ChannelTurn( + channel_type="whatsapp", + session_key="whatsapp:agentA:direct:p1", + conversation_id="whatsapp:p1", + agent="agentA", + user_text="\n\n", + renderer=_Renderer(), + approval_mode="auto", + inbound_route=InboundRoute(conversation_id="p1", text="", attachments_dropped=1), + ) + + asyncio.run(drive_turn(turn, sessions=_Sessions(), ctx_builder=_CtxBuilder())) + + rows = _entries(spool_home) + assert rows[0]["text"] == "" and rows[0]["attachments_dropped"] == 1 + assert "private rules" not in json.dumps(rows) + + +def test_weixin_drive_takes_the_pre_ingestion_originals() -> None: + """Ingestion destroys both values, so ``_drive`` cannot recover them itself. + + ``_ingest_or_refuse`` clears ``inbound.attachments`` and rewrites the text with + temp paths. Reading either at the dispatch site therefore reported ZERO + dropped attachments for a media message and would have quoted on-disk temp + paths into the notice. + """ + import inspect + + from kiro_crew.weixin.transport_dispatch import WeixinDispatcher + + parameters = inspect.signature(WeixinDispatcher._drive).parameters + assert "original_text" in parameters + assert "original_attachments" in parameters + + +def test_weixin_route_text_has_no_fallback_to_the_ingested_prompt() -> None: + """``original_text`` only: the ingested form inlines temp paths.""" + import inspect + + from kiro_crew.weixin.transport_dispatch import WeixinDispatcher + + source = inspect.getsource(WeixinDispatcher._drive) + assert "text=original_text," in source + assert "original_text or text" not in source + + +def test_no_transport_still_carries_a_replay_hook() -> None: + """Re-dispatch was removed on purpose; a stray override would be dead code + reviewers keep re-deriving gates for.""" + from kiro_crew.discord.transport import DiscordTransport + from kiro_crew.messaging.transport import MessagingTransport + from kiro_crew.telegram.transport import TelegramTransport + from kiro_crew.weixin.transport import WeixinTransport + from kiro_crew.whatsapp.transport import WhatsAppTransport + + for cls in ( + MessagingTransport, + TelegramTransport, + DiscordTransport, + WeixinTransport, + WhatsAppTransport, + ): + assert not hasattr(cls, "replay_inbound"), cls.__name__ + assert not hasattr(S, "ReplayOutcome") + assert not hasattr(S, "conversation_moved_on") + assert S.__all__ == ["InboundRoute", "replay_spooled", "spool_refused_turn"] diff --git a/test/test_telegram.py b/test/test_telegram.py index 5a889544c1f..b4dcd985171 100644 --- a/test/test_telegram.py +++ b/test/test_telegram.py @@ -2741,6 +2741,61 @@ async def _go() -> None: # Refused is not leaked -- the session-keyed semaphore still comes back. assert sess.released == ["telegram:kirocrew:direct:7"] + def test_a_shutdown_refusal_is_spooled_for_a_persistent_session( + self, tmp_path, monkeypatch + ) -> None: + """The durable inbound spool (#2217) receives the refused message.""" + from kiro_crew.messaging import inbound_spool as S + + monkeypatch.setattr(S, "data_home", lambda: tmp_path) + d, _cli, sess = _dispatcher({7}) + sess.closing = True + + async def _go() -> None: + await d.handle_message( + InboundMessage( + channel_type="telegram", user_id="7", conversation_id="7", text="keep me" + ) + ) + + asyncio.run(_go()) + + spool = tmp_path / "inbound-spool" / "refused.jsonl" + assert spool.exists() and "keep me" in spool.read_text(encoding="utf-8") + + def test_a_shutdown_refusal_is_not_spooled_for_a_restricted_session( + self, tmp_path, monkeypatch + ) -> None: + """``/incognito`` is a promise that nothing persists, and the spool is a file. + + RED-BEFORE: without the restricted-session gate at the refusal point the + private message is written verbatim to ``refused.jsonl``. The same + predicate that gates the durable-history write gates this one. + """ + from kiro_crew.messaging import inbound_spool as S + + monkeypatch.setattr(S, "data_home", lambda: tmp_path) + d, _cli, sess = _dispatcher({7}) + sess.closing = True + + async def _restricted(_key: str) -> bool: + return True + + monkeypatch.setattr(d, "_session_restricted", _restricted) + + async def _go() -> None: + await d.handle_message( + InboundMessage( + channel_type="telegram", user_id="7", conversation_id="7", text="my secret" + ) + ) + + asyncio.run(_go()) + + spool = tmp_path / "inbound-spool" / "refused.jsonl" + assert not spool.exists(), "an incognito message was persisted to the spool" + assert sess.released == ["telegram:kirocrew:direct:7"], "the refusal must still release" + def test_agent_resolves_to_kirocrew_when_unset(self) -> None: # agent=None + empty default_agent must fall back to "kirocrew" so the # session loads kirocrew-core (spawn_run), not kiro-cli's bare default. diff --git a/test/test_whatsapp_dispatch.py b/test/test_whatsapp_dispatch.py index 86b18a348cb..a5c8cd64fc7 100644 --- a/test/test_whatsapp_dispatch.py +++ b/test/test_whatsapp_dispatch.py @@ -213,6 +213,7 @@ def __init__(self, fail: bool = False, is_operator: bool = True) -> None: self.pending_verdicts: dict[int, GroupVerdict] = {} self.group_gate = FakeGroupGate() self.pending_message_id: dict[int, str] = {} + self.pending_original: dict[int, tuple[str, int]] = {} #: Phase reactions go through the TRANSPORT, not the client: it owns the #: echo tracker, because a reaction is a message and echoes back. self.reactions: list[tuple[str, str]] = [] @@ -643,6 +644,51 @@ async def fake_drive_turn(turn, **kwargs): return seen.get("turn") +# ── durable inbound spool route (#2217) ────────────────────────────────────── + + +def test_dm_route_spools_the_pre_ingestion_original_not_the_prompt(monkeypatch): + """The route text is what the user SENT, read from ``pending_original``. + + By the time the dispatcher runs, ``inbound.text`` has been rewritten by + ``receive`` with attachment context and temp paths, and ``user_text`` may + carry the group's private rules. The restart notice quotes the spooled text, + so only the pre-ingestion original may be spooled. + """ + d, _client, _sessions, transport = _make() + inbound = _msg("look at this\n\n/tmp/kc-att/img-1.jpg") + inbound.attachments = ["/tmp/kc-att/img-1.jpg"] + transport.pending_original[id(inbound)] = ("look at this", 1) + + turn = _captured_turn(monkeypatch, d, inbound) + + assert turn is not None and turn.inbound_route is not None + assert turn.inbound_route.text == "look at this", "the ingested prompt was spooled" + assert turn.inbound_route.attachments_dropped == 1 + assert turn.inbound_route.conversation_id == _DM + + +def test_dm_route_is_not_declared_without_a_captured_original(monkeypatch): + """No fallback to ``inbound.text``: an envelope that skipped ``receive`` is not spooled.""" + d, _client, _sessions, _transport = _make() + + turn = _captured_turn(monkeypatch, d, _msg("hi")) + + assert turn is not None and turn.inbound_route is None + + +def test_group_route_is_never_declared(monkeypatch): + """``may_send_to`` knows nothing of the group roster, so groups are not spooled (#9144).""" + d, _client, _sessions, transport = _make() + inbound = _msg("hi group", conv=_GROUP) + transport.pending_original[id(inbound)] = ("hi group", 0) + transport.pending_verdicts[id(inbound)] = GroupVerdict(respond=True) + + turn = _captured_turn(monkeypatch, d, inbound) + + assert turn is not None and turn.inbound_route is None + + def test_a_non_operator_turn_never_inherits_auto_approval(monkeypatch): """`auto` is the operator's grant, not an admitted stranger's. diff --git a/test/test_whatsapp_transport.py b/test/test_whatsapp_transport.py index 3768eb0447e..7637e497bde 100644 --- a/test/test_whatsapp_transport.py +++ b/test/test_whatsapp_transport.py @@ -525,3 +525,52 @@ async def test_the_operator_can_still_address_a_configured_group(self): event(chat=GROUP, sender=OWN_JID, from_me=True, is_group=True, text="what is next?") ) assert len(h.dispatched) == 1 + + +@pytest.mark.asyncio +class TestPreIngestionOriginalIsCapturedForTheSpool: + """The durable inbound spool (#2217) quotes the spooled text back to the user. + + ``receive`` rewrites ``msg.text`` with attachment context and temp paths + before dispatch, so a route built from ``inbound.text`` at the dispatch site + would spool -- and the restart notice would quote -- on-disk paths to files + that no longer exist. The original caption and media count are captured + BEFORE ingestion in a side table keyed like the others. + """ + + async def test_the_original_caption_survives_ingestion(self, harness, monkeypatch): + import kiro_crew.whatsapp.transport as mod + from kiro_crew.messaging.attachments import IngestResult + + async def fake_ingest(*a, **kw): + return IngestResult(image_paths=["/tmp/kc-att/img-1.jpg"]) + + monkeypatch.setattr(mod, "ingest_media", fake_ingest) + seen: list[tuple[str, tuple[str, int] | None]] = [] + + async def dispatch(msg): + seen.append((msg.text, harness.transport.pending_original.get(id(msg)))) + + harness.transport._dispatch = dispatch + await harness.transport.receive( + event(chat=OWN_JID, sender=OWN_JID, from_me=True, text="look at this", image=True) + ) + + assert len(seen) == 1 + ingested_text, original = seen[0] + assert "/tmp/kc-att/img-1.jpg" in ingested_text, "ingestion did not rewrite the text" + assert original == ("look at this", 1), "the pre-ingestion original was not captured" + assert harness.transport.pending_original == {}, "the side table leaked past dispatch" + + async def test_a_text_only_message_records_zero_media(self, harness): + seen: list[tuple[str, int] | None] = [] + + async def dispatch(msg): + seen.append(harness.transport.pending_original.get(id(msg))) + + harness.transport._dispatch = dispatch + await harness.transport.receive( + event(chat=OWN_JID, sender=OWN_JID, from_me=True, text="just words") + ) + + assert seen == [("just words", 0)]