diff --git a/docs/feature-map/README.md b/docs/feature-map/README.md index de1c1cca1de..a5a1c095ffc 100644 --- a/docs/feature-map/README.md +++ b/docs/feature-map/README.md @@ -81,7 +81,7 @@ this area is reached from inside it unless stated otherwise. | Terminal panel | Two shells on the gateway host: an app-wide docked PTY, and a per-chat terminal whose tab lives in that chat's panel state (opens on the chat's working dir; switches with the session) | Header terminal toggle (docked); chat right panel → **+** menu → **Terminal** (per-chat) | `components/BottomTerminalPanel.tsx`, `pages/chat/SidePanel.tsx` | `handlers/terminal.py` | `POST /api/terminal/sessions`, `GET /api/ws/terminal/{session_id}` | | Browser panel | Live in-panel browser the agent drives | Right panel → **Browser** | `components/WebPreviewPanel.tsx` | `handlers/messaging.py` | `GET,POST /api/browser/view`, `POST /api/browser/command`, `POST /api/browser/command-result` | | Notifications | Bell feed of agent-pushed notifications | Topbar bell → `/notifications` | `pages/NotificationsPage.tsx` | `handlers/messaging.py`, `handlers/notifications_push.py` | `GET /api/notifications`, `POST /api/notifications/ack`, `POST /api/notifications/push` | -| Crew Members | One durable pinned DM thread per crew member; the detail drawer lists the worker sessions the member is driving (live `slots` frames filtered on `created_by`) | `/members` — rail row when the crew preview is on | `pages/members/MembersPage.tsx` | `handlers/members.py`, `slot_projection.py` (`created_by`) | `GET /api/members`, `POST /api/members/{slug}/thread`, `GET /api/members/{slug}/activity`, `GET,PUT /api/members/{slug}/rules`, `GET /api/ws` (`slots`) | +| Crew Members | One durable pinned DM thread per crew member; the detail drawer lists the worker sessions the member is driving (live `slots` frames filtered on `created_by`) | `/members` — rail row when the crew preview is on | `pages/members/MembersPage.tsx` | `handlers/members.py`, `slot_projection.py` (`created_by`), `session_control.py` (`escalate_to_user`, the human as a peer), `src/kiro_crew/crew_conversation.py` (the per-member conversation index) | `GET /api/members`, `POST /api/members/{slug}/thread`, `GET /api/members/{slug}/activity`, `GET,PUT /api/members/{slug}/rules`, `GET /api/members/{slug}/conversation`, `GET /api/ws` (`slots`), `POST /api/session-control/escalate` | | Channels | Group rooms with several agents in one thread | `/channels` (builtin app surface) | `pages/ChannelPage.tsx` | `handlers_channel.py` | `GET,POST /api/channels`, `POST /api/channels/{id}/messages`, `POST /api/channels/{id}/agents` | The Notifications surface is registered `hiddenFromNav`: its route and badge diff --git a/docs/system-specs/modules/README.md b/docs/system-specs/modules/README.md index 9a337e65b6b..8620e05243f 100644 --- a/docs/system-specs/modules/README.md +++ b/docs/system-specs/modules/README.md @@ -23,6 +23,7 @@ agent loads only the one it needs. | [file-search.md](file-search.md) | The `@`-mention file/folder search: index, ranking, `kinds` filter, and the sensitive-path symmetry. | | [session-storage.md](session-storage.md) | What sessions cost on disk, and the user-initiated trash that reclaims it. | | [session-control.md](session-control.md) | One chat session opening, stopping, and reading another. | +| [crew-conversation.md](crew-conversation.md) | The thin per-(human × member) conversation index: pointers, escalation lifecycle, derived `needs_you`. | | [config.md](config.md) | The config schema, defaults, loading, and live reload. | | [cli.md](cli.md) | Every CLI command, the gateway flags, and the test harness. | | [heartbeat.md](heartbeat.md) | The liveness heartbeat and its restricted tool allowlist. | diff --git a/docs/system-specs/modules/crew-conversation.md b/docs/system-specs/modules/crew-conversation.md new file mode 100644 index 00000000000..13e22b3f6a2 --- /dev/null +++ b/docs/system-specs/modules/crew-conversation.md @@ -0,0 +1,113 @@ +# Crew Conversation Module + +## Overview + +A crew member's DM thread on the Crew Members page is a **conversation** +between one human and one member. Its lifetime is longer than any one session: +the pinned DM slot can be rebuilt or re-bound, and a worker session the member +dispatched may hand a result back into it. The conversation therefore has an +identity of its own — but it is deliberately **not** a second transcript. + +`kiro_crew.crew_conversation` keeps that identity thin. The index stores +**pointers and lifecycle**, never bodies: + +| Entry type | Shape | Meaning | +|------------|-------|---------| +| `escalation` | `{id, session_key, mid, from_session, state, created_ts, deadline, default_action, goal, options, answered_ts}` | One `session_escalate` delivery and where it stands. The text lives on the transcript row the pointer names. | + +`state` ∈ `pending | answered | expired | defaulted | retracted`. The record also carries +`participants` (`{kind: human}` + `{kind: member, slug, name}`) and `sessions` +(every session key the conversation spans). The key is `dm:` today; a +later multi-member `goal:` conversation is a new key shape and a longer +`participants` list, not a schema migration — which is why the record is shaped +as lists rather than a `member` / `session` pair. + +## Storage + +`$KIROCREW_HOME/members//conversation.json`, beside `activity.jsonl`, +written whole with `atomic_write` under a per-slug lock (every writer is a +thread of the one gateway process; the lock is what keeps two concurrent +read-modify-writes from dropping each other's entry). It is small: entries +are capped at 500 (~100 KiB), evicting the oldest *settled* entries first — a +pending escalation is never dropped by the cap. It is **not** the trust binding (`trust/member-bindings/.json`): +the binding is the identity authority, strict-shape and keystone-gated; this is +mutable UI state and must stay out of that subtree. An unreadable or missing +file reads as an empty scaffold — the index is derived state and never fails a +roster, a slots frame, or an append. + +`read_conversation` always parses (callers mutate what they get back, so no +shared record is ever cached). The hot path is different: `needs_you` runs +inside the slot projection on every sidebar push for every member slot, **on the +event loop**, so it reads an **in-memory** pending view (`_PENDING_CACHE`: the +pending records' ids and deadlines) and never touches the filesystem — not even +a `stat`. The view is primed off-loop once per member (`prime`, under the same +per-slug lock the writers hold) and refreshed by every writer after its own +write; a member that was never primed reads as nothing pending until a writer or +a prime touches it. The slug → cache-key mapping is bound to the raw +`KIROCREW_HOME` it was resolved under, so a data-home change reads as unprimed +rather than as another home's view. + +## Derived state, not stored state + +`needs_you` is **derived** from the pending escalation records at read time, +projected on the member slot's `slots` frame (`slot_projection.to_dict`) and on +the `GET /api/members` roster row (`needs_you`, `pending_escalations`). It is +never written to the slot: the slot is a process; the conversation is the thing +the human is in. + +Lifecycle transitions: + +- **pending → answered** — a *live* `user` row appended to the member DM slot + (`_ChatSlot.append`, `role == "user"`, `broadcast=True`, `mode == "member"`). + Which record it answers is one rule, shared with the chat projection that + draws the card: a row carrying `meta.escalation_id` (an option chip) answers + exactly that record; a row without one (typed text) answers the pending + record only when exactly one is pending — with none or several it answers + nothing, so an unrelated message cannot retire N open decisions. A replayed + row (`broadcast=False`: transcript rotation, fork, transfer) answers nothing. + A thread with nothing to change costs no write on an ordinary turn. +- **pending → defaulted / expired** — a passed `deadline` is applied lazily on + every read (`sweep_deadlines`): `defaulted` when a `default_action` was + declared (the member proceeds on it), `expired` otherwise. The file is only + rewritten the next time something else writes it (an answer, a new record), + so a deadline passing while the gateway is down still reads correctly on + restart. +- **pending → retracted / answered (recovery)** — the transcript is the truth + and the index a projection of it, so on restore the projection is re-derived + from the transcript (`reconcile_with_transcript`, run by the roster read + until nothing is deferred): a pending record whose card row the transcript + does not hold and that is older than `ORPHAN_GRACE_SECS` (120 s) is an + orphan — the gateway exited between the index write and the slot's flush — + and moves to `retracted` with `retracted_reason: orphan`; a pending record + whose card row is present and that a later durable `user` row with + `meta.human_reply` answers under the live rule moves to `answered` — the + gateway exited between the reply's save and the live hook's index write. A + record still inside the grace is deferred and reconsidered on the next read. + A record whose append failed outright is removed on the spot instead + (`retract_escalation`). + +There is no background poller: nothing needs to *fire* at the deadline, because +the member that set it is the one that acts on it (it stated the default), and +the human-facing card counts down client-side from `meta.deadline`. + +## Read surface + +`GET /api/members/{slug}/conversation` (owner-only, like the thread endpoint; +app tokens refused) returns `public_view(record)`: the swept index plus +`needs_you` / `pending_escalations`. The chat projection on the Crew Members +page does not depend on it — it derives card state from the DM slot's own rows +(a later `user` row = answered; `now > deadline` = expired) so the view can never +disagree with the transcript it is rendered from. + +## What is deliberately not here + +- **No bodies.** A conversation never stores message text; a body lives in + exactly one place, the session JSONL, and is reached through `(session_key, + mid)`. `mid` is minted once (`history.mint_row_mid`) and survives restore, + which is what makes the pointer stable. +- **No per-message refs for the DM slot.** The DM session *is* the + conversation's main body; indexing every row of it would duplicate the + transcript's own ordering for nothing. +- **No approval records.** An escalation is a decision the member may take on + its own if unanswered; approvals (which block) are a different object and + are not modelled here. diff --git a/docs/system-specs/modules/session-control.md b/docs/system-specs/modules/session-control.md index f85010890c9..b36dd9b2143 100644 --- a/docs/system-specs/modules/session-control.md +++ b/docs/system-specs/modules/session-control.md @@ -20,6 +20,7 @@ unreachable in production because the caller's `X-Internal-Secret` is ignored. | `session_stop` | `POST /api/session-control/stop` | Stop another session's in-flight turn | | `session_close` | `POST /api/session-control/close` | Close (archive) another session, as the tab ✕ does — heavier than stop, and recoverable rather than a delete | | `session_send` | `POST /api/session-control/send` | Deliver a message that another session runs as its next turn | +| `session_escalate` | `POST /api/session-control/escalate` | Raise something to the human who owns the caller, as a peer — a card in the owning member's DM thread, no turn started (see below) | | `session_read_message` | `GET /api/session-control/read` | Read another session's transcript tail + liveness | **One verb here writes into another session's conversation: `session_send`.** @@ -226,10 +227,12 @@ injection would strip a member thread of its tools mid-conversation. On the KAS backend the wire agent projection additionally grants the server in `tools` plus the member's approval-free dashboard verbs in `allowedTools` (ceiling-filtered like every other grant): `_MEMBER_DASHBOARD_GRANTS`, the -conductor's read/create set plus `session_send` and `session_stop` — the +conductor's read/create set plus `session_send`, `session_stop` and +`session_escalate` — the write verbs are safe to auto-approve for a member *specifically* because the `created_by` ownership fence above bounds them to worker sessions the member -itself opened. Member sessions also bypass the provider warm pool +itself opened, and the escalation verb because it writes only into the member's +own thread. Member sessions also bypass the provider warm pool (`bypass_member`): a pooled child was spawned with no session key on the default backend, so a warm hit would skip both the member backend route and the mount. The member backend is `agent.member_acp_backend` (default `kas`), @@ -513,6 +516,251 @@ identity not presence" discipline `create_session` uses for its slot allocation, and the same theme as the queued-drain re-check (#5911). The human ✕ path passes no check — the person owns the tab and closes it unconditionally. +## Escalating to the human (`session_escalate`) + +The human is addressable as a peer, through a verb of its own. `session_escalate` +does not start a turn anywhere; it lands one `escalation` row in the DM thread of +the crew member that owns the caller, rings the bell, and returns. Where the card +lands is fixed by who is calling: + +| Caller | Lands in | Member index touched | +|--------|----------|----------------------| +| A member DM slot (`member-`) | its own thread | yes | +| A worker session a member created (`_created_by` is a member slot) | the **creating member's** thread | yes | +| Any other session | its own transcript | no | + +It is deliberately **not** a reserved `target` of `session_send`. The two differ +in kind — a send delivers text the target session *runs as a turn*; an +escalation writes a row and runs nothing — and a shared name would have merged +two parameter sets into one schema (four fields silently dead unless `target` +happened to be one literal) and, worse, made their policy inseparable: every +site that classifies tools keys on the **name**. Each of those sites therefore +names `session_escalate` on its own, and the decision recorded at each is: + +| Site | Decision | Why | +|------|----------|-----| +| `SESSION_CONTROL_TOOLS` (`mcp_dashboard.py`) | in | the caller's verified identity is the authorization, as for every other verb here | +| `CHANNEL_AGENT_BLOCKED_TOOLS` (`channel.py`) | **blocked** | not on `session_send`'s grounds (nothing runs) but on `send_notification`'s: the card is mirrored onto the notification bus, which is exactly the reach-the-user path that list closes, and its row lands in a transcript the channel agent's own conversation is not. The backend agrees — a channel-linked or mirrored caller is refused (`linked_session_caller` / `mirrored_caller`). Letting a channel agent ask its human is a two-site change (this entry plus that gate), made on purpose or not at all | +| `_MEMBER_DASHBOARD_GRANTS` (`agent.py`) | granted | it runs nothing and writes only into the member's own thread, and a member that hits a wall mid-turn with nobody at the keyboard is the case the verb exists for; the conductors keep it mounted-but-gated like every other write | +| `MCP_DASHBOARD_SCHEMAS` / the registration pin | own schema (`message`, `deadline?`, `default_action?`, `options?`, `goal?`, no `target`) | `session_send` is back to `target` + `message`; an escalation field on it is refused as an unknown field, never dropped | + +A session may be titled `user`; nothing here consults titles, so there is no +collision to refuse. + +The human is not a slot, so none of the *target* gates apply — but every +*caller* gate of `authorize_target` does, in the same order and with the same +codes (`caller_unidentified`, `session_control_disabled` with the member +bypass, `unattended_caller`, `caller_gone`, `app_scoped_caller`, +`ephemeral_caller`, `linked_session_caller`, `mirrored_caller`): an escalation +is still a session reaching outside its own transcript. For a worker the +result says `reply_in_caller_thread: false` — the human answers in the +member's thread, and the worker learns of it only if the member relays it +(the member is the actor that owns the worker, not the human). + +The row is `role: escalation`, `cls: msg msg-escalation`, content the caller's +markdown (one line of background, what was tried, what is needed — the tool +description asks for exactly that shape), and `meta`: +`{kind: "escalation", escalation_id, from_session, deadline, default_action, +options, goal, state: "pending", created_ts, mid}`. `deadline` is normalised to +an absolute ISO timestamp from a duration (`30m`, `2h`, `1d`) or an ISO input +and must fall 1 minute to 7 days out; `options` is at most six de-duplicated +strings of at most 120 characters; `default_action` and `goal` are at most 500. +Every field is sanitised with the same `sanitize_outbound` the peer path uses, +because the text crosses from one session into a transcript the human reads. +An invalid field is refused (`400`, `deadline_invalid` / `options_too_many` / +…) before anything is appended. + +**`options` and the two other choice surfaces.** Three mechanisms put a choice +in front of the human, and they must not answer each other: + +1. The `[OPTIONS: a | b]` trailing marker on an assistant turn, parsed into + follow-up pills by `deriveFollowUpOptions` (`website/src/app-sdk/protocol/options.ts`). +2. The `ask_question` card (`questionPending`), which belongs to the **local** + session's own turn. +3. Escalation `options`, carried on the row's `meta.options` and the index + record. + +The rules: an `escalation` row **ends** the pill scan — it offers no pills and, +crucially, does not let the scan walk past it to a previous turn's marker, which +is how a stale chip could otherwise post as a live `user` row that the one-pending +rule below reads as the answer to an escalation it never matched. The row is +**claimed by the default renderer registry** (`EscalationNotice`, `pages/chat`), +so every surface that shows the thread — the Crew Members thread the bell +deep-links to is a `ChatPane` — draws the question: the member's markdown, the +veto window **in words** ("Unless you reply by , the member will: +"), the offered options as **readable text**, and how to answer (type in +the thread). The interactive card (option chips, live countdown, state badge — +`chatProfileRenderers` / `EscalationCard`, #8614) replaces that entry by claiming +the role and answers with `meta.escalation_id`; until it lands the human answers +by typing (the free-text rule). The bell mirror's body says the same things in +the same words (a UTC clock time, no glyphs). The tool description promises +exactly that and nothing more. +An `ask_question` card is resolved through its own endpoint and writes **no** +transcript row, so answering it never answers an escalation, and an escalation +arriving while a card is open changes nothing about the card: the card holds +the local turn, the escalation is a row from another session, and each is +answered on its own surface. (The card's 404 fallback — resending the answer as +an ordinary composer message — is an ordinary `user` row and follows the +free-text rule like any other typed message.) + +A member's escalation is also recorded on that member's **conversation index** +(see [crew-conversation.md](crew-conversation.md)): a pending record pointing +at `(session_key, mid)`, written *before* the card is surfaced under a +pre-minted row id so a reply can never race the record. `needs_you` — +projected on the member slot's `slots` frame and on the `GET /api/members` +roster row — is *derived* from the pending records, never stored on the slot. + +Which reply answers which record is one rule, applied identically by the +index (`mark_answered`) and by the chat projection that draws the card: + +- a live `user` row carrying `meta.escalation_id` (an option chip) answers + exactly that record — the id rides the busy paths too (a queue entry's meta, + a steer row's meta), because a member that just escalated is usually + mid-turn when the human answers; when the queue drain merges several + entries into one row, only an entry that is *itself* the human's + (`human_reply`) contributes its id (`escalation_ids`), and the merged row is + the human's only if **every** merged entry was — one automated or non-owner + entry in the batch and the row answers nothing; a **mixed** batch (chips and + typed text) is replayed **in order** at the drain, where the order is still + known, against the pending view — a chip answers its record and takes it + off the table, typed text answers the single record left at its position or + nothing — and the row then *names* every record the sequence answered, so + the index, the recovery replay and the projection all apply their unchanged + named-id rule (with no pending view available the row names the chips + alone: the under-answering side); +- a live `user` row without one (typed text) answers the pending record only + when **exactly one** was pending *at the moment the row was appended* — the + hook snapshots that set from the index's in-memory view on the event loop, + so a record landing on the executor a beat later is neither counted nor + answered, and the index agrees with transcript order; with none or several + pending it answers nothing, so an unrelated message cannot silently retire N + open decisions; +- a record whose deadline has passed is swept to `defaulted` (a + `default_action` was declared) or `expired` first and is never answered + late — judged against the **reply row's own timestamp**, not the moment of + the index write, so a timely reply whose durable save crosses the deadline + still answers (and `answered_ts` is the row's); a replayed row (fork, + rotation, transfer) answers nothing; only a row + the **authenticated composer** produced answers at all — the chat handler + stamps `meta.human_reply` server-side only on the auth layer's positive + `is_dashboard_user` signal (a validated dashboard credential; app tokens, + derived internal callers and the internal secret never qualify — a falsy + `app` claim is not trust) **and** the owner identity + (`is_owner_dashboard_request`: the member's DM thread is the owner's + conversation, so a non-owner dashboard user who can post there is not the + human the escalation was raised to), never trusted from the client, and carries it + through the queue, steer and requeue paths, so a peer's `session_send` row, + a heartbeat, a cron `prompt:` or an internal caller posting into the member + slot, which all land as `user` rows, never answer. + +The index record is written *before* the card under a pre-minted row id, and +the write is load-bearing: if it fails the escalation is refused +(`escalation_index_unavailable`, 500) rather than delivered without a +lifecycle. The index also holds a **ceiling on open decisions per member** +(`MAX_PENDING_ESCALATIONS`, 50): pending records are never evicted by the file's +size cap, so a member escalating in a loop with nobody answering is refused at +the ceiling (`escalation_backlog_full`, 429) — atomically under the slug lock, +before any row exists, with the count in the message — and told to wait for +answers or report to its owner. Only genuinely open records count: a passed +deadline is swept first and an answered record frees a slot. The card row is +then appended and surfaced like any other row and +persisted by the slot's ordinary flush — there is no forced save and no +rollback on this path. If the thread closes, the caller loses eligibility, or +the worker changes workspace across the index await, the record is retracted +and the call refused (`target_gone`, 409); if the append itself fails, the +record is retracted. A worker whose creating member's thread is in another +workspace is refused up front (`workspace_mismatch`), the same boundary the +peer path holds. + +**Recovery: reconcile with the transcript on restore.** Consistency between +the index and the transcript runs in one direction — the transcript is the +truth, the index a thin projection of it — so on restore the projection is +re-derived from the transcript, both ways. An index file that exists but cannot +be read (a torn write, a hand edit, the wrong shape) is not an empty index: +writers refuse to overwrite it (`IndexUnreadable`; an escalation is refused as +`escalation_index_unavailable`, a live answer-mark is skipped and logged), the +roster forces a reconciliation regardless of the transcript's generation, and +the reconciliation **rebuilds** every card row the transcript holds as a real +pending record from the row's own meta before replaying replies and sweeping +deadlines — the file is repaired from the truth rather than replaced by a blank. Before a member's pending count is +trusted, each roster read in a gateway process hands the transcript in order +(the persisted rows — including the rotated archive, since the restored live +window is only a tail — followed by live rows not yet flushed, which count for +**card presence only**: their `human_reply` provenance is stripped before the +replay, because settling a record on the strength of a reply that is not yet on +disk would let a gateway exit before the flush restore an answered index with +no reply in the transcript; the live hook, which persists first, is what +answers a live reply) to +`crew_conversation.reconcile_with_transcript`: a pending record older than +`ORPHAN_GRACE_SECS` (120 s) whose card row is absent (the gateway exited +between the index write and the slot's flush) moves to `retracted` +(`retracted_reason: orphan`); a pending record whose card row is present and +that a later `user` row with `meta.human_reply` answers under the live rule +(named `escalation_id`/`escalation_ids`, or free text with exactly one pending +at that point of the transcript) moves to `answered` (the gateway exited +between the reply's transcript save and the live hook's index write). Replies +are replayed *before* deadlines are swept, each judged against the record's +deadline as of the reply's own row timestamp, so a timely reply is never +recorded as `defaulted` because recovery ran after the deadline. A record +still inside the grace is *deferred* — neither retracted nor trusted — and the +member is marked reconciled only when nothing was deferred, so it is looked at +again on the next read. The mark is keyed to the transcript's generation +(persisted mtime plus live-window length), not to the process: a rewind or any +rewrite that removes a card row changes the generation and the index is +re-derived on the next read whether or not the member has anything pending (a +rewind can drop the reply that **answered** a record while its card stays: that +record is provisionally reopened and must find its reply again in the replay — +one that does not is reopened for real, `reopened` in the result, then judged +against its deadline like any other; an answered record whose card row is gone +is left as history). A card the transcript still holds but the index no longer +does — the size cap evicts settled entries oldest-first while their rows live on +— takes part in the replay as a **counting-only** candidate, so a typed reply +that was ambiguous when it was made stays ambiguous and cannot falsely +re-answer a reopened record; such phantoms are never written back. The live hook keeps the same +direction: the human's reply row is persisted first and the record is marked +`answered` only when that save committed; an unprimed memory view is not an +empty one — the hook falls back to the file rather than treating "nothing +cached" as "nothing pending". The hook's persist-and-mark tasks are **serialised +per slot in transcript order** (each awaits its predecessor), so a chip for A +appended just before a typed reply has marked A before the typed reply is +judged — otherwise a slow save for the chip let the typed reply run first, see +two pending, decline, and leave B lit until the next reply. + +The reply that answers also pushes a slots update, so the badge clears with +the reply. One SEL `session_escalate` line is written per delivery, naming the +escalation id, the target thread, and whether a deadline, default and options +were carried; caller-side refusals are audited as `session_control.escalate` +denials with the same codes the peer path uses. + +Three constraints are the contract, and each has a corresponding refusal or +absence in the code rather than a note in the tool description: + +1. **Non-blocking, with a veto window.** Delivery *is* success; the caller's turn + continues and nothing awaits the human. A caller that can proceed on a + sensible default states `deadline` + `default_action` — "unless you stop me + by *t*, I do *X*" — and the window closing is recorded as `defaulted`, never + as a failure. The human's reply is an ordinary user turn in the member + thread, with the option chips on the card sending their text as that turn. + There is no `session_wait_for_reply`; a caller that wants the answer polls + its own thread with `session_read_message` like any other input. +2. **Attention budget.** The card's home is the member's DM thread, where the + chat projection folds escalations by `goal`. Anything that mirrors it + elsewhere goes through the notification bus — `system.agent`, `kind: + escalation`, `group_key: escalation::` — so N escalations on one + goal stack as one entry in the bell today and route as one item once the + per-channel "deliver to additional channels" bridge (RFC notification + bridge) exists. There is deliberately **no transport-specific switch** on + this path: a `slack_escalate`-style boolean would be a persisted contract + the bridge would have to migrate on day one. +3. **Escalation is not approval.** This verb carries a decision the member is + allowed to make on its own if unanswered. Anything the human must + *actively grant* — objective and metric definitions, budgets, the member's + own continuation or scope — must not travel here, and this PR implements no + approval flow. The distinction is not blocking-vs-non-blocking + (`ask_question` is itself non-blocking) but *who may act if nobody answers*: + an escalation lets the member proceed on its stated default; an approval + never does, and stays on the approval surfaces that wait for the grant. + ## Configuration `agent.session_control` (bool, default **true**). The grant that decides who may @@ -555,10 +803,21 @@ folder tools individually. into another session's conversation, but only one the same `authorize_target` guard admits: a channel-linked, channel-mirrored, crew-mode, incognito, app-scoped, unattended or cross-workspace target is refused, so the verb cannot - reach a conversation other people are party to. The residual is the queued arm's + reach a conversation other people are party to. `session_escalate` reaches + only the caller's own thread or its creating member's — never + a third session. The residual is the queued arm's second authorization moment, recorded above and tracked as #5911. - **No cross-workspace or cross-machine reach.** The boundary is one gateway's live sessions in one workspace. +- **No deadline wakeup.** Nothing fires when an escalation's veto window + closes: the index reads the record as `defaulted`/`expired` on the next + read, and the member that declared the default is the one that acts on it + — a member whose turn ends before its own deadline must arrange its own + wake (a monitor loop or a schedule); the tool description and the success + reply both say so, because that member is the only actor who can keep the + promise the card made. `defaulted` therefore records that the window closed + with a default declared — not that the action ran. A timer here would make + the gateway, not the member, the actor of record for the default. - **No waking closed sessions.** See above. - **No writes on the read path.** `session_read_message` never changes the target's state, so a poll loop cannot perturb what it is measuring. diff --git a/src/kiro_crew/agent.py b/src/kiro_crew/agent.py index 0fab4a1dcc9..d9bd87bb42a 100644 --- a/src/kiro_crew/agent.py +++ b/src/kiro_crew/agent.py @@ -5320,10 +5320,15 @@ def _install_research_agent() -> None: #: such ownership fence, which is why its list withholds the writes. Without #: these two the dispatch loop this feature exists for (create → seed → patrol #: → stop) stalls on an approval prompt at its second step with nobody at the -#: keyboard. +#: keyboard. ``session_escalate`` joins on a narrower bound still: it runs +#: nothing and writes only into the member's OWN DM thread (``_escalation_home`` +#: routes a member caller to itself), so its worst case is a card and a bell in +#: the thread the human already reads — and a member that hits a wall mid-turn +#: with nobody at the keyboard is exactly when it must not stall on approval. _MEMBER_DASHBOARD_GRANTS: tuple[str, ...] = _CONDUCTOR_DASHBOARD_GRANTS + ( "@kirocrew-dashboard/session_send", "@kirocrew-dashboard/session_stop", + "@kirocrew-dashboard/session_escalate", ) diff --git a/src/kiro_crew/channel.py b/src/kiro_crew/channel.py index 2119dbe66e9..7f6d5aa9279 100644 --- a/src/kiro_crew/channel.py +++ b/src/kiro_crew/channel.py @@ -55,6 +55,15 @@ # only cancels and read only exfiltrates, but send delivers text that the target # session RUNS as a turn — so external channel content would execute inside a # private dashboard conversation. +# ESCALATE runs nothing, so SEND's reasoning does not carry over; it is blocked +# on send_notification's grounds instead: the card is mirrored onto the +# notification bus (bell badge and sound), which is exactly the reach-the-user +# path this list closes, and its row lands in a transcript the channel agent's +# own conversation is not. The backend agrees -- a channel-linked or mirrored +# caller is refused by ``_authorize_escalation_caller`` (``linked_session_caller`` +# / ``mirrored_caller``) -- so opening escalation to channel agents is a +# deliberate two-site change (this entry plus that gate), not an accident of +# a shared name. # Matched against the rendered # permission-request text/title via _blocked_tool_named() (boundary-aware, # not naive substring — "Editing send_notification.py" must NOT match). @@ -63,6 +72,7 @@ "send_notification", "session_stop", "session_send", + "session_escalate", "session_read_message", "session_create", "session_close", diff --git a/src/kiro_crew/crew_conversation.py b/src/kiro_crew/crew_conversation.py new file mode 100644 index 00000000000..6e5ecbc359d --- /dev/null +++ b/src/kiro_crew/crew_conversation.py @@ -0,0 +1,974 @@ +"""Crew conversation index — the thin per-(human × member) conversation entity. + +A crew member's DM thread on the Crew Members page is a *conversation* between +one human and one member. Its lifetime is longer than any single session: the +DM slot can be rebuilt, rotated, or re-bound, and a worker session the member +dispatched may hand a result back into it. The conversation therefore needs an +identity of its own — but it must NOT become a second transcript. + +This module keeps that identity **thin**: + +* it stores **pointers**, never bodies — an entry is either a + ``(session_key, mid)`` reference into a session's JSONL transcript, or a + native *escalation* record whose text still lives on the transcript row; +* the human-facing projection (what the chat view shows) is computed from the + referenced transcripts, so a conversation can never disagree with the + sessions it points at; +* ``needs_you`` is **derived** from the pending escalations on the index, not + stored on the slot — the slot is a process, the conversation is the thing the + human is in. + +The key is ``dm:`` today (one human, one member). The record already +carries a ``participants`` list and a ``sessions`` list rather than a single +member/session field, so a later ``goal:`` conversation (one goal, several +members plus the human) is a new key shape, not a schema migration. + +Same placement discipline as the activity log (:mod:`kiro_crew.members`): +the file sits in the member's own directory, beside ``activity.jsonl``, and is +NOT the trust binding — the binding is the identity authority and stays +strict-shape; this is mutable UI state and stays out of the keystone-gated +subtree. +""" + +from __future__ import annotations + +import json +import logging +import os +import re +import threading +import uuid +from collections.abc import Mapping, Sequence +from datetime import datetime, timedelta, timezone +from pathlib import Path +from typing import Any + +from kiro_crew.atomic_write import atomic_write +from kiro_crew.members import member_dir, validate_slug + +logger = logging.getLogger(__name__) + +#: File name inside ``member_dir(slug)``. +CONVERSATION_FILE_NAME = "conversation.json" + +SCHEMA_VERSION = 1 + +#: Cap on SETTLED entries per conversation — pointers are ~200 bytes, so the +#: settled history stays around 100 KiB. Eviction is oldest-first over settled +#: entries only: a pending escalation is never dropped by the cap (a badge that +#: vanished without an answer, a deadline or a default would be a lost +#: decision, not a trimmed log), so a record with more than this many OPEN +#: decisions is allowed to exceed the cap. +_MAX_ENTRIES = 500 + +#: Hard ceiling on OPEN decisions per member. Pending records are never evicted +#: by the cap above, so without this a member escalating deadline-free in a +#: loop, with nobody answering, would grow the file without bound — every +#: write rewrites it whole. Past the ceiling ``record_escalation`` REFUSES +#: (``EscalationBacklogFull``), atomically under the slug lock, and the caller +#: surfaces that to the member: the fix for a hundred unanswered decisions is +#: an answer or a stop, not a hundred-and-first card. Generous enough that a +#: human's backlog never hits it; a runaway loop hits it in seconds. +MAX_PENDING_ESCALATIONS = 50 + + +class EscalationBacklogFull(RuntimeError): + """Raised by :func:`record_escalation` when the member already has + :data:`MAX_PENDING_ESCALATIONS` open decisions.""" + + def __init__(self, slug: str, pending: int) -> None: + super().__init__( + f"{slug} already has {pending} unanswered escalations " + f"(limit {MAX_PENDING_ESCALATIONS})" + ) + self.slug = slug + self.pending = pending + + +#: Cap on option labels an escalation may offer (mirrors ``ask_question``). +MAX_ESCALATION_OPTIONS = 6 + +#: Transcript role of an escalation card row. Spelled here as well as in +#: ``session_control.ESCALATION_ROLE`` (which imports this module, so the +#: dependency cannot run the other way); ``test_escalation`` pins the two equal. +ESCALATION_ROW_ROLE = "escalation" + +# One lock per slug around every read-modify-write. All writers live in one +# gateway process (the dashboard's own executor threads: the escalation path, +# the reply hook), but they run on different threads, so +# without this two concurrent mutations would each load the file, each append +# their entry and the second `atomic_write` would silently drop the first. +_LOCKS: dict[str, threading.Lock] = {} +_LOCKS_GUARD = threading.Lock() + +# In-memory pending view for the hot read path (`needs_you` runs inside the +# slot projection, ON the event loop, on every sidebar push): index file path +# -> (id, deadline) of the records stored as pending. Never read from disk on +# the read path; the writers refresh it after every write and `prime` loads it +# once per member. +_PENDING_CACHE: dict[str, list[tuple[str, str | None]]] = {} + +# slug -> (the raw ``KIROCREW_HOME`` value it was resolved under, its resolved +# index-file cache key). The read path (`pending_ids`, `is_primed`) runs ON the +# event loop and must never resolve a filesystem path there (`member_dir` +# calls `.resolve()`, which stats/readlinks and can freeze the loop on a +# stalled filesystem). Only the OFF-LOOP writers and `prime` compute the key +# (`_resolve_cache_key`), recording it here; the on-loop reader looks it up by +# slug with no IO. A data-home change is handled without IO too: the raw env +# value is memoized with the key, and a lookup under a different value is +# treated as unresolved — so the reader reads "nothing pending" until the next +# off-loop writer/prime recomputes the key, never another home's view. +_CACHE_KEY_BY_SLUG: dict[str, tuple[str | None, str]] = {} + + +def _raw_home() -> str | None: + """The raw ``KIROCREW_HOME`` value — an environment read, no filesystem.""" + return os.environ.get("KIROCREW_HOME") + + +def _resolve_cache_key(slug: str) -> str: + """Resolve *slug*'s index-file cache key AND memoize it by slug. Touches + the filesystem (`.resolve()` via `conversation_path`), so call it only + OFF the event loop — the writers and `prime` already run off-loop.""" + key = str(conversation_path(slug)) + _CACHE_KEY_BY_SLUG[slug] = (_raw_home(), key) + return key + + +def _memo_cache_key(slug: str) -> str | None: + """The slug's cache key from memory, or ``None`` if no off-loop path has + resolved it yet — or resolved it under a different ``KIROCREW_HOME``, in + which case the mapping is stale and must not project that home's view. + No filesystem access — safe on the event loop.""" + memo = _CACHE_KEY_BY_SLUG.get(slug) + if memo is None or memo[0] != _raw_home(): + return None + return memo[1] + + +def _lock_for(slug: str) -> threading.Lock: + with _LOCKS_GUARD: + lock = _LOCKS.get(slug) + if lock is None: + lock = _LOCKS[slug] = threading.Lock() + return lock + + +def conversation_id(slug: str) -> str: + """The conversation key for a member's 1:1 DM with the human.""" + return f"dm:{validate_slug(slug)}" + + +def conversation_path(slug: str) -> Path: + """Absolute path of one member's conversation index (not created).""" + return member_dir(slug) / CONVERSATION_FILE_NAME + + +def _now_iso(now: datetime | None = None) -> str: + return (now or datetime.now(timezone.utc)).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def parse_ts(value: Any) -> datetime | None: + """Parse an ISO-8601 timestamp (``Z`` or offset) to an aware datetime.""" + if not isinstance(value, str) or not value: + return None + text = value.strip() + if text.endswith("Z") or text.endswith("z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed + + +_DURATION_UNITS = {"s": 1, "m": 60, "h": 3600, "d": 86400} +#: Bounds on a relative deadline: below a minute the veto window is not real; +#: above a week the escalation should have been a decision, not a window. +MIN_DEADLINE_SECS = 60 +MAX_DEADLINE_SECS = 7 * 86400 + + +def resolve_deadline(value: Any, *, now: datetime | None = None) -> str | None: + """Normalise a caller-supplied deadline to an absolute ISO timestamp. + + Accepts an ISO-8601 timestamp, a bare number of seconds, or a duration + such as ``30m`` / ``2h`` / ``900s`` / ``1d``. Returns ``None`` for an + empty value. Raises :class:`ValueError` for anything unparseable or a + window outside ``[MIN_DEADLINE_SECS, MAX_DEADLINE_SECS]`` from *now*. + """ + if value is None: + return None + base = now or datetime.now(timezone.utc) + if isinstance(value, bool): + raise ValueError("deadline must be a timestamp or a duration") + if isinstance(value, (int, float)): + secs = float(value) + else: + text = str(value).strip() + if not text: + return None + absolute = parse_ts(text) + if absolute is not None: + secs = (absolute - base).total_seconds() + else: + unit = text[-1].lower() + number = text[:-1].strip() + if unit in _DURATION_UNITS and number.replace(".", "", 1).isdigit(): + secs = float(number) * _DURATION_UNITS[unit] + elif text.replace(".", "", 1).isdigit(): + secs = float(text) + else: + raise ValueError("deadline must be ISO-8601 or a duration like 30m / 2h / 900s") + if secs < MIN_DEADLINE_SECS or secs > MAX_DEADLINE_SECS: + raise ValueError( + f"deadline must be between {MIN_DEADLINE_SECS}s and {MAX_DEADLINE_SECS // 86400}d from now" + ) + return _now_iso(base + timedelta(seconds=secs)) + + +#: The one spelling of an escalation id. Boundaries that accept an id from a +#: client (the chat send handler's ``meta.escalation_id``) validate against this +#: rather than trusting free text into a queue entry. +ESCALATION_ID_RE = re.compile(r"^esc-[0-9a-f]{16}$") + + +def new_escalation_id() -> str: + """Mint one escalation id (``esc-<16 hex>``); random for the same reason + ``history.mint_row_mid`` is — a counter rebased after restore can collide.""" + return f"esc-{uuid.uuid4().hex[:16]}" + + +def _scaffold(slug: str) -> dict[str, Any]: + return { + "version": SCHEMA_VERSION, + "conversation_id": conversation_id(slug), + "participants": [], + "sessions": [], + "entries": [], + } + + +def _parse_file(path: Path, slug: str) -> dict[str, Any] | None: + """Parse the index file, or ``None`` when missing/unreadable/misshapen.""" + try: + data = json.loads(path.read_text(encoding="utf-8")) + except OSError: + return None + except ValueError: + logger.warning("conversation index unreadable for %s", slug, exc_info=True) + return None + if not isinstance(data, dict) or not isinstance(data.get("entries"), list): + return None + record = _scaffold(slug) + # Type-check every field a writer later mutates: a hand-edited or torn file + # with ``"participants": null`` must read as "no participants", not crash the + # next ``record_escalation`` in ``_ensure_participants``. + if isinstance(data.get("conversation_id"), str) and data["conversation_id"]: + record["conversation_id"] = data["conversation_id"] + if isinstance(data.get("version"), int): + record["version"] = data["version"] + record["participants"] = ( + [p for p in (data.get("participants") or []) if isinstance(p, dict)] + if isinstance(data.get("participants"), list) + else [] + ) + record["sessions"] = ( + [s for s in (data.get("sessions") or []) if isinstance(s, str)] + if isinstance(data.get("sessions"), list) + else [] + ) + record["entries"] = [e for e in data["entries"] if isinstance(e, dict)] + return record + + +class IndexUnreadable(RuntimeError): + """The index file EXISTS but cannot be read as an index (torn write, hand + edit, wrong shape). Writers refuse rather than overwrite it — the transcript + still holds every card, and the next roster read rebuilds the index from it + (:func:`reconcile_with_transcript`); overwriting would destroy what the + rebuild needs to distinguish.""" + + +def _load_index(slug: str) -> tuple[dict[str, Any], str]: + """``(record, state)`` where state is ``"missing"`` (no file: the scaffold + IS the truth), ``"ok"`` (parsed), or ``"unreadable"`` (a file is there but + not an index; the scaffold stands in for reading only).""" + path = conversation_path(slug) + if not path.exists(): + return _scaffold(slug), "missing" + parsed = _parse_file(path, slug) + if parsed is None: + return _scaffold(slug), "unreadable" + return parsed, "ok" + + +def read_conversation(slug: str) -> dict[str, Any]: + """Load a member's conversation index; a missing or unreadable file reads + as an empty scaffold (never raises — the index is derived state). + + Always parses: callers mutate what they get back, so no shared cached + record is ever handed out. The hot path (:func:`needs_you`) has its own + scalar cache. + """ + return _load_index(slug)[0] + + +def index_unreadable(slug: str) -> bool: + """Whether an index file exists for *slug* but cannot be read. The roster + uses this to force a reconciliation regardless of the transcript's + generation: a corrupted index is a change the transcript's mtime does not + record.""" + return _load_index(slug)[1] == "unreadable" + + +def _load_for_write(slug: str) -> dict[str, Any]: + """The record a WRITER may mutate. Caller holds the slug lock. Refuses an + unreadable file (see :class:`IndexUnreadable`): a writer that started from + the scaffold would overwrite the only copy of a possibly-recoverable + lifecycle; a missing file is simply the first write.""" + record, state = _load_index(slug) + if state == "unreadable": + raise IndexUnreadable(f"conversation index for {slug} exists but cannot be read") + return record + + +def _settled(entry: dict[str, Any]) -> bool: + return not (entry.get("type") == "escalation" and entry.get("state") == "pending") + + +def _write_conversation(slug: str, record: dict[str, Any]) -> None: + """Persist *record*. Caller holds the slug lock.""" + path = conversation_path(slug) + path.parent.mkdir(parents=True, exist_ok=True) + entries = record["entries"] + overflow = len(entries) - _MAX_ENTRIES + if overflow > 0: + # Evict the oldest SETTLED entries only. A pending escalation is never + # evicted: if every entry is an open decision the file simply exceeds + # the cap, because a lost decision is worse than a large file (500 + # unanswered escalations is a member that needs stopping, not trimming). + keep: list[dict[str, Any]] = [] + for entry in entries: + if overflow > 0 and _settled(entry): + overflow -= 1 + continue + keep.append(entry) + record["entries"] = keep + # The sessions list is a pointer set over the entries; once an entry is + # gone, a session nothing points at is dropped with it so the record's + # size is bounded by the entries, not by history. + referenced = {e.get("session_key") for e in keep} | { + e.get("from_session") for e in keep if e.get("type") == "escalation" + } + record["sessions"] = [s for s in record.get("sessions", []) if s in referenced] + atomic_write(path, json.dumps(record, ensure_ascii=False, indent=1), fsync=False) + _PENDING_CACHE.pop(_resolve_cache_key(slug), None) + _prime_pending_cache(slug) + + +def _ensure_participants(record: dict[str, Any], *, member: str, slug: str) -> None: + parts = record.setdefault("participants", []) + if not any(p.get("kind") == "human" for p in parts if isinstance(p, dict)): + parts.append({"kind": "human", "id": "owner"}) + if not any( + p.get("kind") == "member" and p.get("slug") == slug for p in parts if isinstance(p, dict) + ): + parts.append({"kind": "member", "slug": slug, "name": member}) + + +def _ensure_session(record: dict[str, Any], session_key: str) -> None: + sessions = record.setdefault("sessions", []) + if session_key and session_key not in sessions: + sessions.append(session_key) + + +def record_escalation( + slug: str, + *, + member: str, + session_key: str, + mid: str, + escalation_id: str, + from_session: str, + created_ts: str = "", + deadline: str | None = None, + default_action: str | None = None, + goal: str | None = None, + options: list[str] | None = None, +) -> dict[str, Any]: + """Add one pending escalation record pointing at its transcript row.""" + entry = { + "type": "escalation", + "id": escalation_id, + "session_key": session_key, + "mid": mid, + "from_session": from_session, + "state": "pending", + "created_ts": created_ts or _now_iso(), + "deadline": deadline, + "default_action": default_action, + "goal": goal, + "options": list(options or [])[:MAX_ESCALATION_OPTIONS], + "answered_ts": None, + } + with _lock_for(slug): + record = _load_for_write(slug) + # Count what is genuinely open: a passed deadline is settled first so a + # backlog of defaulted/expired records never blocks a live escalation. + # The check and the append happen under the same lock, so two racing + # escalations cannot both squeeze in at the ceiling. + sweep_deadlines(record) + open_count = sum( + 1 + for e in record.get("entries", []) + if e.get("type") == "escalation" and e.get("state") == "pending" + ) + if open_count >= MAX_PENDING_ESCALATIONS: + raise EscalationBacklogFull(slug, open_count) + _ensure_participants(record, member=member, slug=slug) + _ensure_session(record, session_key) + record["entries"].append(entry) + _write_conversation(slug, record) + return entry + + +def sweep_deadlines(record: dict[str, Any], *, now: datetime | None = None) -> bool: + """Move pending records whose deadline has passed to ``defaulted`` / + ``expired`` in place. Pure over *record*; returns whether anything moved.""" + current = now or datetime.now(timezone.utc) + changed = False + for entry in record.get("entries", []): + if entry.get("type") != "escalation" or entry.get("state") != "pending": + continue + due = parse_ts(entry.get("deadline")) + if due is not None and due <= current: + entry["state"] = "defaulted" if entry.get("default_action") else "expired" + changed = True + return changed + + +def pending_escalations(record: dict[str, Any], *, now: datetime | None = None) -> list[dict]: + """Escalations still awaiting the human, after a lazy deadline sweep.""" + sweep_deadlines(record, now=now) + return [ + e + for e in record.get("entries", []) + if e.get("type") == "escalation" and e.get("state") == "pending" + ] + + +def _pending_deadlines_from_disk(slug: str) -> list[tuple[str, str | None]]: + """``(id, deadline)`` of the records stored as ``pending`` (unswept), read + from disk. Blocking IO — callers run it off the event loop.""" + record = _parse_file(conversation_path(slug), slug) + if record is None: + return [] + return [ + ( + str(e.get("id") or ""), + e.get("deadline") if isinstance(e.get("deadline"), str) else None, + ) + for e in record["entries"] + if e.get("type") == "escalation" and e.get("state") == "pending" + ] + + +def pending_ids(slug: str, *, now: datetime | None = None) -> list[str]: + """Ids of the escalations awaiting the human RIGHT NOW, from the in-memory + view (no IO). The reply hook snapshots this on the event loop at the moment + the human's row is appended, so the answer rule is evaluated against the + pending set as it stood in transcript order — not as it stands a moment + later on the executor, after a concurrent escalation may have landed.""" + current = now or datetime.now(timezone.utc) + out: list[str] = [] + key = _memo_cache_key(slug) + if key is None: + return out # never resolved off-loop: unprimed reads as nothing pending + for eid, deadline in _PENDING_CACHE.get(key, ()): + due = parse_ts(deadline) + if due is None or due > current: + out.append(eid) + return out + + +def _prime_pending_cache(slug: str) -> None: + """Refresh the in-memory pending view for *slug* from disk (blocking IO).""" + try: + _PENDING_CACHE[_resolve_cache_key(slug)] = _pending_deadlines_from_disk(slug) + except Exception: # noqa: BLE001 - a cache refresh must never raise into a writer + logger.debug("pending cache prime failed for %s", slug, exc_info=True) + + +def prime(slug: str) -> None: + """Load a member's pending view into memory. Blocking IO — call it off the + event loop (``asyncio.to_thread``) once per member at slot creation or + restore; every later change is applied by the writer that made it. + + Serialised on the same per-slug lock every writer holds across its write + AND its cache refresh: an unlocked prime that read the file just before a + writer's ``atomic_write`` would install the OLD pending view a beat after + the writer installed the new one, and a free-text reply judged against that + stale view would leave the fresh escalation pending. (``_prime_pending_cache`` + itself stays lock-free because ``_write_conversation`` calls it while the + lock is already held — ``threading.Lock`` is not re-entrant.)""" + with _lock_for(slug): + _prime_pending_cache(slug) + + +def needs_you(slug: str, *, now: datetime | None = None) -> bool: + """Whether the member has at least one escalation awaiting the human. + + Memory-only: this runs inside the slot projection on the event loop, on + every sidebar push, so it must not stat or read a file. The view is kept + current by the writers (``record_escalation`` and ``mark_answered`` + both refresh it after their write) and primed by + :func:`prime` when a member slot is created or restored. A slug that was + never primed reads as ``False`` until a writer or a prime touches it. + + A passed deadline clears it without a write (the file is updated the next + time the record is written for another reason, or by :func:`mark_answered`). + """ + try: + return bool(pending_ids(slug, now=now)) + except Exception: # noqa: BLE001 - a projection must never fail on derived state + logger.debug("needs_you derivation failed for %s", slug, exc_info=True) + return False + + +def mark_answered( + slug: str, + *, + escalation_id: str | None = None, + escalation_ids: list[str] | None = None, + candidates: list[str] | None = None, + answered_ts: str = "", + now: datetime | None = None, +) -> int: + """The human replied in the conversation. Which record that answers: + + * a reply carrying an ``escalation_id`` (an option chip) answers exactly + that record, if it is still pending; a reply carrying several + (``escalation_ids`` — chip replies merged into one row by the queue + drain) answers each of them; + * a reply without one (typed text) answers the pending record only when + EXACTLY ONE is pending — with none or several it answers nothing, so an + unrelated message cannot silently retire N open decisions; + * a record whose deadline has already passed is swept to + ``defaulted``/``expired`` first and is never answered late. + + ``candidates`` is the set of ids that were pending when the reply row was + appended (:func:`pending_ids`, snapshotted on the event loop). It is what + the free-text rule counts, so a record that landed on the executor between + the append and this call is neither counted nor answered — the index then + agrees with the transcript order the chat projection reads. Without a + snapshot the rule falls back to the records pending now. + + The chat projection applies the same rule client-side, so the card and the + index agree without a round trip. Also persists any deadline transitions + found on the way. Returns the number of records moved to ``answered``; a + conversation with nothing to change is left untouched (no write). + """ + with _lock_for(slug): + record = _load_for_write(slug) + swept = sweep_deadlines(record, now=now) + pending = [ + e + for e in record.get("entries", []) + if e.get("type") == "escalation" and e.get("state") == "pending" + ] + if candidates is not None: + allowed = set(candidates) + pending = [e for e in pending if e.get("id") in allowed] + else: + # Unprimed fallback (no on-loop snapshot): stand in transcript + # order by hand. A record CREATED after the reply row — a second + # escalation that landed on the executor during the forced save — + # was not pending when the human replied, so the free-text rule + # must not count it: otherwise the reply that should answer the + # one open record sees two pending and answers neither, leaving the + # first permanently pending. Mirrors ``reconcile_with_transcript`` + # (pending AT that point of the transcript). A record with no + # ``created_ts`` predates this field and is kept. + reply_cutoff = now or datetime.now(timezone.utc) + + def _created_at_or_before(entry: dict[str, Any]) -> bool: + made = parse_ts(entry.get("created_ts")) + return made is None or made <= reply_cutoff + + pending = [e for e in pending if _created_at_or_before(e)] + targets: list[dict[str, Any]] + named = [ + i for i in ([escalation_id] if escalation_id else []) + list(escalation_ids or []) if i + ] + if named: + wanted = set(named) + targets = [e for e in pending if e.get("id") in wanted] + elif len(pending) == 1: + targets = pending + else: + targets = [] + stamp = answered_ts or _now_iso(now) + for entry in targets: + entry["state"] = "answered" + entry["answered_ts"] = stamp + if targets or swept: + _write_conversation(slug, record) + return len(targets) + + +def retract_escalation(slug: str, escalation_id: str) -> bool: + """Remove a record whose transcript row never materialised. + + The escalation path writes the index BEFORE it surfaces the card (so a fast + reply cannot race the record); if the append then fails, this is the + compensation — without it a no-deadline record would keep ``needs_you`` lit + for a card nobody can see. Returns whether a record was removed. + """ + with _lock_for(slug): + record, state = _load_index(slug) + if state == "unreadable": + return False # nothing legible to retract from; never overwrite it + before = len(record["entries"]) + record["entries"] = [ + e + for e in record["entries"] + if not (e.get("type") == "escalation" and e.get("id") == escalation_id) + ] + if len(record["entries"]) == before: + return False + _write_conversation(slug, record) + return True + + +#: How old a pending record must be before a missing transcript row makes it an +#: orphan. The escalation path writes the index BEFORE it appends the card, and +#: the append is a loop hop away from that write; a record younger than this is +#: presumed in flight, never swept — it is *deferred*, and the caller keeps +#: coming back until nothing is deferred. +ORPHAN_GRACE_SECS = 120 + + +def _row_named_ids(meta: Any) -> list[str]: + if not isinstance(meta, dict): + return [] + out: list[str] = [] + one = meta.get("escalation_id") + if isinstance(one, str) and one: + out.append(one) + many = meta.get("escalation_ids") + if isinstance(many, list): + out.extend(i for i in many if isinstance(i, str) and i) + return out + + +def reconcile_with_transcript( + slug: str, + rows: Sequence[Mapping[str, Any]], + *, + now: datetime | None = None, + session_key: str = "", + member: str = "", +) -> dict[str, Any]: + """Recovery: make the index agree with the member's transcript, both ways. + + Consistency between this index and the transcript runs in one direction — + the transcript is the truth, the index a thin projection of it — so on + restore the projection is re-derived from *rows* (the transcript in order, + persisted rows plus any live rows not yet flushed). When the index is + missing or unreadable, the projection is REBUILT: every card row becomes a + pending record again (``session_key``/``member`` name the thread it lives + in) before the replay below answers what the transcript answers. + + * **orphans** — a pending record whose card row is not in *rows* and that is + older than :data:`ORPHAN_GRACE_SECS` (the gateway exited between the index + write and the slot's flush) moves to ``retracted`` (``retracted_reason: + orphan``). A younger one is *deferred*: its append may simply not have + happened yet. + * **durable answers** — a pending record whose card row IS present and that + a later ``user`` row with ``meta.human_reply`` answers under the same rule + :func:`mark_answered` applies live (a named ``escalation_id`` / + ``escalation_ids`` answers those records; free text answers the record + only when it is the ONLY one pending at that point of the transcript) + moves to ``answered`` — covering a gateway exit between the reply's + transcript save and the live hook's index write. + + Replies are replayed first, each judged against the deadline as of its own + row timestamp; only then are still-unresolved deadlines swept. Returns ``{"retracted": + [...], "answered": [...], "deferred": n}``; the caller treats ``deferred > + 0`` as "not done yet" and reconciles again on its next read. + """ + current = now or datetime.now(timezone.utc) + position: dict[str, int] = {} + for i, row in enumerate(rows): + mid = (row.get("meta") or {}).get("mid") if isinstance(row.get("meta"), dict) else None + if isinstance(mid, str) and mid and mid not in position: + position[mid] = i + with _lock_for(slug): + record, index_state = _load_index(slug) + # Order matters: transcript replies are replayed FIRST, each judged + # against its record's deadline as of the reply's own timestamp, and + # only then are the deadlines of whatever is still unresolved swept. A + # timely reply the crash kept out of the index must never be recorded as + # ``defaulted`` because recovery happened to run after the deadline. + changed = False + retracted: list[str] = [] + deferred = 0 + pending: list[dict[str, Any]] = [] + rebuilt: list[str] = [] + # The index is a projection of the transcript. When there is NO usable + # projection — the file is unreadable (torn write, hand edit) or gone + # while the transcript still holds cards — rebuild every card as a REAL + # pending record from the row's own meta, then let the replay below + # answer and the sweep settle them exactly as it would have. Writers + # refuse an unreadable file (``_load_for_write``), so this rebuild is + # what repairs it. With a readable index, a card it does not hold is + # an evicted SETTLED one and is only a counting phantom (below). + index_is_truth = index_state == "ok" + if not index_is_truth: + for row in rows: + if row.get("role") != ESCALATION_ROW_ROLE: + continue + rmeta = row.get("meta") + if not isinstance(rmeta, dict): + continue + eid = rmeta.get("escalation_id") + mid = rmeta.get("mid") + if not (isinstance(eid, str) and eid and isinstance(mid, str) and mid in position): + continue + if any(e.get("id") == eid for e in record["entries"]): + continue + stamp = row.get("ts") if isinstance(row.get("ts"), str) and row.get("ts") else None + raw_options_val = rmeta.get("options") + raw_options: list[Any] = ( + raw_options_val if isinstance(raw_options_val, list) else [] + ) + entry = { + "type": "escalation", + "id": eid, + "session_key": session_key, + "mid": mid, + "from_session": ( + rmeta.get("from_session") + if isinstance(rmeta.get("from_session"), str) + else "" + ), + "state": "pending", + "created_ts": ( + rmeta.get("created_ts") + if isinstance(rmeta.get("created_ts"), str) and rmeta.get("created_ts") + else (stamp or _now_iso(current)) + ), + "deadline": ( + rmeta.get("deadline") if isinstance(rmeta.get("deadline"), str) else None + ), + "default_action": ( + rmeta.get("default_action") + if isinstance(rmeta.get("default_action"), str) + else None + ), + "goal": rmeta.get("goal") if isinstance(rmeta.get("goal"), str) else None, + "options": [o for o in raw_options if isinstance(o, str)][ + :MAX_ESCALATION_OPTIONS + ], + "answered_ts": None, + } + record["entries"].append(entry) + if member or session_key: + _ensure_participants(record, member=member or slug, slug=slug) + _ensure_session(record, session_key) + rebuilt.append(eid) + changed = True + if rebuilt: + logger.warning( + "conversation index for %s was %s; rebuilt %d escalation record(s) " + "from the transcript", + slug, + index_state, + len(rebuilt), + ) + # ``answered`` records whose card row is still in the transcript are + # re-derived too: the reply that answered them may have been REWOUND + # (a rewind rewrites the transcript, never this index). They are + # provisionally reopened and must find their reply again in the replay + # below; one that does not is genuinely open once more. Keyed by id so a + # reply that re-answers with the same stamp leaves the record untouched. + provisional: dict[str, tuple[str, Any]] = {} + for entry in record.get("entries", []): + if entry.get("type") != "escalation": + continue + state = entry.get("state") + if state == "answered" and entry.get("mid") in position: + provisional[str(entry.get("id", ""))] = ("answered", entry.get("answered_ts")) + entry["state"] = "pending" + pending.append(entry) + continue + if state != "pending": + continue + if entry.get("mid") in position: + pending.append(entry) + continue + created = parse_ts(entry.get("created_ts")) + if created is not None and (current - created).total_seconds() < ORPHAN_GRACE_SECS: + deferred += 1 + continue + entry["state"] = "retracted" + entry["retracted_reason"] = "orphan" + retracted.append(str(entry.get("id", ""))) + changed = True + # Cards the transcript still holds but the index no longer does: the + # size cap evicts SETTLED entries oldest-first while their card rows + # live on. They were candidates at their positions when the human + # replied, so the replay must see them too — otherwise a typed reply + # that was ambiguous (two open cards) reads as unambiguous against the + # one card the index kept and falsely answers a reopened record. They + # take part in candidate counting only: phantoms are never written. + known_ids = { + str(e.get("id", "")) for e in record.get("entries", []) if e.get("type") == "escalation" + } + for row in rows: + if row.get("role") != ESCALATION_ROW_ROLE: + continue + rmeta = row.get("meta") + if not isinstance(rmeta, dict): + continue + eid = rmeta.get("escalation_id") + mid = rmeta.get("mid") + if not (isinstance(eid, str) and eid and eid not in known_ids): + continue + if not (isinstance(mid, str) and mid in position): + continue + known_ids.add(eid) + pending.append( + { + "type": "escalation", + "id": eid, + "mid": mid, + "state": "pending", + "deadline": rmeta.get("deadline"), + "_phantom": True, + } + ) + answered: list[str] = [] + if pending: + for i, row in enumerate(rows): + if row.get("role") != "user": + continue + meta = row.get("meta") + if not isinstance(meta, dict) or meta.get("human_reply") is not True: + continue + stamp = row.get("ts") if isinstance(row.get("ts"), str) and row.get("ts") else None + at = parse_ts(stamp) or current + # Pending at THAT point of the transcript, and not yet past its + # deadline as of the reply — a late reply answers nothing, the + # same rule the live path applies. + candidates = [] + for e in pending: + if e.get("state") != "pending" or position[e["mid"]] >= i: + continue + due = parse_ts(e.get("deadline")) + if due is not None and due <= at: + continue + candidates.append(e) + if not candidates: + continue + named = set(_row_named_ids(meta)) + if named: + targets = [e for e in candidates if e.get("id") in named] + elif len(candidates) == 1: + targets = candidates + else: + targets = [] + for entry in targets: + new_ts = stamp or _now_iso(current) + entry["state"] = "answered" + entry["answered_ts"] = new_ts + if entry.get("_phantom"): + continue # counted as a candidate; never part of the record + eid = str(entry.get("id", "")) + if provisional.get(eid) == ("answered", new_ts): + continue # the same reply, still there: nothing changed + answered.append(eid) + changed = True + reopened: list[str] = [] + for entry in pending: + eid = str(entry.get("id", "")) + if eid in provisional and entry.get("state") == "pending": + # Was answered; the transcript no longer holds the reply. + entry["answered_ts"] = None + reopened.append(eid) + changed = True + if sweep_deadlines(record, now=current): + changed = True + if changed: + _write_conversation(slug, record) + return { + "retracted": retracted, + "answered": answered, + "reopened": reopened, + "rebuilt": rebuilt, + "deferred": deferred, + } + + +def is_primed(slug: str) -> bool: + """Whether :func:`prime` (or a writer) has loaded *slug*'s pending view into + memory. An unprimed view is not an EMPTY view: a reader that would treat + "nothing cached" as "nothing pending" must fall back to the file instead.""" + key = _memo_cache_key(slug) + return key is not None and key in _PENDING_CACHE + + +def public_view(record: dict[str, Any], *, now: datetime | None = None) -> dict[str, Any]: + """The index as the dashboard reads it: swept, with ``needs_you`` derived.""" + pending = pending_escalations(record, now=now) + return { + "conversation_id": record.get("conversation_id", ""), + "participants": list(record.get("participants", [])), + "sessions": list(record.get("sessions", [])), + "entries": list(record.get("entries", [])), + "needs_you": bool(pending), + "pending_escalations": len(pending), + } + + +def invalidate_cache(slug: str | None = None) -> None: + """Test hook / explicit cache drop.""" + if slug is None: + _PENDING_CACHE.clear() + _CACHE_KEY_BY_SLUG.clear() + else: + memo = _CACHE_KEY_BY_SLUG.pop(slug, None) + _PENDING_CACHE.pop(memo[1] if memo else _resolve_cache_key(slug), None) + + +__all__ = [ + "CONVERSATION_FILE_NAME", + "ESCALATION_ID_RE", + "ESCALATION_ROW_ROLE", + "EscalationBacklogFull", + "IndexUnreadable", + "MAX_DEADLINE_SECS", + "MAX_ESCALATION_OPTIONS", + "MAX_PENDING_ESCALATIONS", + "MIN_DEADLINE_SECS", + "ORPHAN_GRACE_SECS", + "conversation_id", + "conversation_path", + "index_unreadable", + "is_primed", + "mark_answered", + "needs_you", + "new_escalation_id", + "parse_ts", + "pending_escalations", + "prime", + "public_view", + "read_conversation", + "reconcile_with_transcript", + "record_escalation", + "resolve_deadline", + "sweep_deadlines", +] diff --git a/src/kiro_crew/dashboard/chat_delivery.py b/src/kiro_crew/dashboard/chat_delivery.py index 2d4cbda71da..d776d4ef7a3 100644 --- a/src/kiro_crew/dashboard/chat_delivery.py +++ b/src/kiro_crew/dashboard/chat_delivery.py @@ -231,6 +231,8 @@ async def steer_into_running_turn( message: str, *, send_id: str | None = None, + escalation_id: str | None = None, + human_reply: bool = False, ) -> str: """Inject *message* into the slot's RUNNING turn; return a ``STEER_*`` outcome. @@ -313,6 +315,15 @@ async def steer_into_running_turn( # requeued entry's meta byte-identical to its pre-#6751 shape. if send_id: slot._steer_send_ids[message] = send_id + if escalation_id: + # Same shape as the two maps above: a steer the turn's teardown requeues + # must not lose which escalation record its reply names. + slot._steer_escalation_ids[message] = escalation_id + if human_reply: + # And the VALIDATED provenance rides the same bookkeeping: the requeue + # must restore exactly what the handler established, never promote an + # internal caller's steer to a human reply. + slot._steer_human_reply.add(message) slot._pending_steers.append(message) try: steered = await client.steer(message) @@ -336,6 +347,8 @@ async def steer_into_running_turn( # every intermediate transition, including a merged row. slot._steer_delivery_ids.pop(message, None) slot._steer_send_ids.pop(message, None) + slot._steer_escalation_ids.pop(message, None) + slot._steer_human_reply.discard(message) logger.info( "steer for slot %s was requeued and drained during the RPC; row already " "persisted", slot.key, @@ -355,6 +368,8 @@ async def steer_into_running_turn( slot._pending_steers.remove(message) slot._steer_delivery_ids.pop(message, None) slot._steer_send_ids.pop(message, None) + slot._steer_escalation_ids.pop(message, None) + slot._steer_human_reply.discard(message) return STEER_UNAVAILABLE if stopped: # Still registered means the teardown has not run yet and will @@ -432,6 +447,8 @@ async def steer_into_running_turn( # few lines below, so nothing will read the map entry again and leaving it # would hold a full message string for the slot's lifetime. slot._steer_send_ids.pop(message, None) + slot._steer_escalation_ids.pop(message, None) + slot._steer_human_reply.discard(message) ts = datetime.now(timezone.utc).isoformat() # Cut the in-flight text segment at the steer boundary BEFORE persisting the @@ -499,6 +516,12 @@ async def steer_into_running_turn( # transcript page is what mergePreservedThinking reads to resolve an # optimistic bubble by id (accepted steer vs raced new turn, #6075). meta["sendId"] = send_id + if escalation_id: + # Same reason as the queue path: a chip reply steered into a running + # member turn must still name the escalation record it answers. + meta["escalation_id"] = escalation_id + if human_reply: + meta["human_reply"] = True # Store the sanitized form — raw content must never reach an external # surface — so the steer survives a page reload via the dirty-flush cycle. _row = slot.append("user", sanitized, "msg msg-u", ts=ts, meta=meta) @@ -534,6 +557,8 @@ def queue_for_next_turn( *, directive_user_origin: bool = False, send_id: str | None = None, + escalation_id: str | None = None, + human_reply: bool = False, ) -> str: """Append *message* to the slot's queue and announce it; return the queue id. @@ -550,6 +575,14 @@ def queue_for_next_turn( back to text, which a same-text resend or an injection can share). Additive: a send whose POST carried no usable id stores nothing here and the entry meta keeps the exact prior shape. + + ``escalation_id`` rides the queue entry's ``meta`` (the drain merges entry + meta onto the user row it appends) so an option-chip reply to a crew + member's escalation still names the record it answers when the member was + mid-turn — the common case, since a member that just escalated is working. + ``human_reply`` rides the same way: the authenticated composer sets it, and + it is what lets the drained row answer an escalation at all (an automated + prompt's row never carries it). """ # circular import: session_control imports this module at module level. from kiro_crew.dashboard.session_control import containment_meta @@ -557,6 +590,10 @@ def queue_for_next_turn( meta: dict[str, Any] = containment_meta(state, slot) if send_id: meta["sendId"] = send_id + if escalation_id: + meta["escalation_id"] = escalation_id + if human_reply: + meta["human_reply"] = True qid = slot.queue_append( message, meta=meta, diff --git a/src/kiro_crew/dashboard/chat_handlers.py b/src/kiro_crew/dashboard/chat_handlers.py index a77b0c219cd..97a392f7d42 100644 --- a/src/kiro_crew/dashboard/chat_handlers.py +++ b/src/kiro_crew/dashboard/chat_handlers.py @@ -37,6 +37,7 @@ published_autocompact_pct, resolve_agent_bindings, ) +from kiro_crew.crew_conversation import ESCALATION_ID_RE as _ESCALATION_ID_RE from kiro_crew.dashboard import remote_mirror from kiro_crew.dashboard.channel_slots import channel_slot_name, note_slot_closed from kiro_crew.dashboard.chat_auto_tag import maybe_auto_tag @@ -208,6 +209,56 @@ def _sweep_stale_permissions(slot: "_ChatSlot") -> None: ) +def _sanitize_escalation_meta( + user_meta: dict | None, +) -> "tuple[dict | None, str | None]": + """Return ``(user_meta, escalation_id)`` with the client's escalation keys + reduced to the ONE validated singular id. + + An option-chip reply names the record it answers via ``meta.escalation_id``, + validated to the id grammar the index mints so a client cannot smuggle + arbitrary text into a row through this key. ``escalation_ids`` (plural) is a + server-internal merge artifact — the queue drain writes it when several + replies fold into one row — and is NEVER accepted from a client: a supplied + list would let one POST name, and answer, every pending decision at once. + An unvalidated ``escalation_id`` is dropped rather than carried as opaque + text. Every other key passes through untouched. + """ + if not user_meta: + return user_meta, None + raw = user_meta.get("escalation_id") + escalation_id = raw if isinstance(raw, str) and _ESCALATION_ID_RE.fullmatch(raw) else None + if "escalation_id" not in user_meta and "escalation_ids" not in user_meta: + return user_meta, None + cleaned = {k: v for k, v in user_meta.items() if k not in ("escalation_id", "escalation_ids")} + if escalation_id: + cleaned["escalation_id"] = escalation_id + return (cleaned or None), escalation_id + + +def _human_reply_provenance(request: Any) -> bool: + """Whether a composer send may carry ``meta.human_reply`` — the provenance + the escalation answer rule keys on. Two positive signals are required: + the auth layer's ``is_dashboard_user`` (a validated dashboard credential — + ``token_auth`` sets it ``True`` there and ``False`` for app tokens and + derived internal callers) AND the owner identity + (``is_owner_dashboard_request``): a crew member's DM thread is the OWNER's + conversation, and a non-owner dashboard user (an allowed Slack user holding + a ``!dashboard`` token) may be able to post into it but is not the human + the escalation was raised to. Absence, ``None`` or a falsy ``app`` claim is + NOT trust (CWE-269): an internal-secret caller posting into a member thread + is automation, and its row must not clear the human's escalation.""" + if request.get("is_dashboard_user") is not True: + return False + # circular import: source_providers imports from this module's package. + from kiro_crew.dashboard.handlers.source_providers import is_owner_dashboard_request + + try: + return bool(is_owner_dashboard_request(request)) + except Exception: # noqa: BLE001 - no owner identity resolvable -> not the owner + return False + + async def api_chat(request: web.Request) -> web.StreamResponse: """POST /api/chat — send message to a slot, stream response via SSE.""" state: DashboardState = request.app["state"] @@ -222,6 +273,25 @@ async def api_chat(request: web.Request) -> web.StreamResponse: user_meta = body.get("meta") # knowledge/files/pastes metadata from frontend if not isinstance(user_meta, dict): user_meta = None + # An option-chip reply to a crew member's escalation names the record it + # answers. Validated to the id grammar the index mints, so a client cannot + # smuggle arbitrary text into a queue entry's meta through this key; carried + # through the busy paths (steer / queue) because the drained user row is + # what the answer rule reads. + user_meta, escalation_id = _sanitize_escalation_meta(user_meta) + # Provenance the escalation answer rule keys on. Only the POSITIVE + # dashboard-user signal the auth layer sets (``is_dashboard_user`` — a + # validated dashboard credential, not an app token and not the internal + # secret) earns the stamp: a falsy ``app`` claim is not trust (CWE-269), + # and an internal caller posting into a member thread is automation, not + # the human. An unstamped user row (heartbeat, cron, peer, internal + # caller) can never answer an escalation. Stamped server-side, never + # trusted from the client: a client-supplied value is overwritten. + human_reply = _human_reply_provenance(request) + if human_reply: + user_meta = {**(user_meta or {}), "human_reply": True} + elif user_meta and "human_reply" in user_meta: + user_meta = {k: v for k, v in user_meta.items() if k != "human_reply"} or None theme_consent = body.get("theme_consent") is True # Content-bound persona consent: the sha256 hex the user # granted in the consent modal. Injection is gated on this matching the @@ -573,6 +643,8 @@ async def api_chat(request: web.Request) -> web.StreamResponse: slot, message, send_id=user_meta.get("sendId") if user_meta else None, + escalation_id=escalation_id, + human_reply=human_reply, ) if outcome == STEER_STEERED: return web.json_response({"ok": True, "steered": True}) @@ -636,6 +708,8 @@ async def api_chat(request: web.Request) -> web.StreamResponse: message, directive_user_origin=not bool(request_app), send_id=normalize_send_id(user_meta.get("sendId")) if user_meta else None, + escalation_id=escalation_id, + human_reply=human_reply, ) return web.json_response({"ok": True, "queued": True, "queue_id": qid}) @@ -693,6 +767,10 @@ async def api_chat(request: web.Request) -> web.StreamResponse: _hold_sid = normalize_send_id(user_meta.get("sendId")) if user_meta else None if _hold_sid: _hold_meta["sendId"] = _hold_sid + if escalation_id: + _hold_meta["escalation_id"] = escalation_id + if human_reply: + _hold_meta["human_reply"] = True qid = slot.queue_append( message, meta=_hold_meta, @@ -3546,6 +3624,10 @@ async def stop_slot_turn( # kill discards the text, so there is no requeued entry left to carry # the client's send id onto. slot._steer_send_ids.pop(_discarded, None) + slot._steer_escalation_ids.pop(_discarded, None) + # And its human-reply provenance: a later same-text automated steer + # must not inherit it and be requeued as the human's answer. + slot._steer_human_reply.discard(_discarded) slot._pending_steers.clear() state.push_slots_update() logger.info("Stop (force): hard-killing session for slot %s", name) diff --git a/src/kiro_crew/dashboard/chat_runner.py b/src/kiro_crew/dashboard/chat_runner.py index 07207fc9072..923af17d503 100644 --- a/src/kiro_crew/dashboard/chat_runner.py +++ b/src/kiro_crew/dashboard/chat_runner.py @@ -4796,6 +4796,18 @@ def _requeue_unconsumed_steers(state: "DashboardState", slot: "_ChatSlot") -> No _sid = getattr(slot, "_steer_send_ids", {}).pop(steer_msg, "") if _sid: _meta["sendId"] = _sid + # And the escalation record an option-chip reply names, for the same + # reason: the drained row is what the answer rule reads. + _eid = getattr(slot, "_steer_escalation_ids", {}).pop(steer_msg, "") + if _eid: + _meta["escalation_id"] = _eid + # The requeued row keeps exactly the provenance the handler validated + # for the original steer — recorded per message like the escalation id + # — so an internal caller's steer is never promoted to a human reply. + _hr = getattr(slot, "_steer_human_reply", None) + if _hr is not None and steer_msg in _hr: + _hr.discard(steer_msg) + _meta["human_reply"] = True # Provenance is derivable, not guessed: `steer_into_running_turn` has # exactly one caller (the api_chat composer branch), and app isolation # confines app-surface requests to app-scoped slots — so every steer @@ -5078,6 +5090,64 @@ def _drop_stale_admissions(state: DashboardState, slot: _ChatSlot) -> None: ) +def _replay_merged_escalation_replies( + slot: _ChatSlot, replies: list[str | None], named: list[str] +) -> list[str]: + """Resolve a MIXED merged batch of human replies (chips and typed text) to + the records it answers, in order. + + *replies* is the human's entries in queue order — an escalation id for a + chip, ``None`` for typed text. Each chip answers its record and removes it + from what is pending; each typed reply answers the single record still + pending at its position, or nothing when zero or several remain (the same + free-text rule ``crew_conversation.mark_answered`` applies to a lone row). + Returns the ids in answer order, *named* first when the pending view is not + available. + + The pending view is the index's in-memory snapshot taken on the loop + (``pending_ids``), the same source the live answer hook reads, so the + resolution and the hook agree on what was open. A slot that is not a + member's DM thread, an unprimed view or a read failure fall back to the + chips alone — the under-answering side, which leaves a badge the human's + next reply clears rather than answering a record with text meant for + another. + """ + from kiro_crew.members import DM_SLOT_KEY_PREFIX + + key = getattr(slot, "key", "") or "" + if not key.startswith(DM_SLOT_KEY_PREFIX): + return list(named) + # circular import: crew_conversation imports members, which sits below this + # module in the layering (members -> artifacts -> validation). + from kiro_crew.crew_conversation import is_primed, pending_ids + + slug = key[len(DM_SLOT_KEY_PREFIX) :] + if not is_primed(slug): + return list(named) + try: + open_ids = set(pending_ids(slug)) + except Exception: # noqa: BLE001 - derived state; fall back to the chips alone + logger.debug("escalation merged-reply replay failed for %s", slug, exc_info=True) + return list(named) + resolved: list[str] = [] + for item in replies: + if item is not None: + if item not in resolved: + resolved.append(item) + open_ids.discard(item) + continue + remaining = [i for i in open_ids if i not in resolved] + if len(remaining) == 1: + resolved.append(remaining[0]) + open_ids.discard(remaining[0]) + # A chip for a record no longer pending still names it (the index ignores + # an already-answered id); never drop a named id the human clicked. + for item in named: + if item not in resolved: + resolved.append(item) + return resolved + + async def _start_next_queued_turn(state: DashboardState, slot: _ChatSlot) -> bool: """Dequeue and start one ready Kiro turn, preserving queue semantics.""" @@ -5318,6 +5388,9 @@ async def _start_next_queued_turn(state: DashboardState, slot: _ChatSlot) -> boo # one) keeps the plain-send shape -- `sendId` alone -- so a client reading # only that key sees exactly what a dispatched send's row carries. _drained_send_ids: list[str] = [] + _drained_escalation_ids: list[str] = [] + _drained_human_replies: list[str | None] = [] + _drained_all_human = True # circular import: session_control imports this package's modules at module level. from kiro_crew.dashboard.session_control import ( QUEUED_CONTAINMENT_META_KEY, @@ -5349,12 +5422,40 @@ async def _start_next_queued_turn(state: DashboardState, slot: _ChatSlot) -> boo _sid = _item_meta.get("sendId") if isinstance(_sid, str) and _sid and _sid not in _drained_send_ids: _drained_send_ids.append(_sid) + # Escalation replies accumulate for the same reason steer ids do: + # two option chips answered while the member was busy become ONE + # drained row, and last-wins merging would lose the first answer. + # Only an entry that is ITSELF the human's (``human_reply``) may + # contribute an id: a non-owner's chip merged with the owner's text + # must not ride out on an owner-marked row and answer a record the + # owner never touched. + _item_human = _item_meta.get("human_reply") is True + _esc = _item_meta.get("escalation_id") + if ( + _item_human + and isinstance(_esc, str) + and _esc + and _esc not in _drained_escalation_ids + ): + _drained_escalation_ids.append(_esc) + if _item_human: + # The human's replies IN ORDER: a chip names its record, typed + # text (None) names nothing and is resolved below against what + # was still pending at its position in the batch. + _drained_human_replies.append(_esc if isinstance(_esc, str) and _esc else None) + if not _item_human: + _drained_all_human = False # The admission-time containment snapshot (#5911) is queue plumbing, # consumed by _drop_stale_admissions above; it says nothing about the # ROW, so it must not ride into the persisted transcript meta. + # ``human_reply`` is decided for the merged row as a whole below. _drained_meta.update( - (k, v) for k, v in _item_meta.items() if k != QUEUED_CONTAINMENT_META_KEY + (k, v) + for k, v in _item_meta.items() + if k not in (QUEUED_CONTAINMENT_META_KEY, "escalation_id", "human_reply") ) + else: + _drained_all_human = False if _drained_ids: _drained_meta.pop("steer_delivery_id", None) _drained_meta["steer_delivery_ids"] = _drained_ids @@ -5364,6 +5465,30 @@ async def _start_next_queued_turn(state: DashboardState, slot: _ChatSlot) -> boo # send the row stands for. Membership in `sendIds` is the proof a client # should read on a merged row; on a plain row `sendId` is the whole story. _drained_meta["sendIds"] = _drained_send_ids + # A merged row is the human's reply only if EVERY merged entry was: one + # automated or non-owner entry in the batch and the row answers nothing. + if consumed and _drained_all_human: + _drained_meta["human_reply"] = True + # A MIXED batch — at least one chip and at least one typed reply merged + # into this row — is replayed in order here, where the order is still + # known, against what the member had pending: a chip answers its record + # and takes it off the table, typed text answers the record only if + # exactly one is left at its position (the free-text rule), else + # nothing. The row then NAMES every record the sequence answered, so + # the index, the recovery replay and the chat projection all apply + # their unchanged named-id rule and agree. Without this, "chip for A, + # then text" merged into one row answered A alone and left B pending + # behind a lit badge, while the same two replies drained separately + # would have answered both. Pure-chip and pure-text batches are + # unchanged: they never had an order problem. + if None in _drained_human_replies and _drained_escalation_ids: + _drained_escalation_ids = _replay_merged_escalation_replies( + slot, _drained_human_replies, _drained_escalation_ids + ) + if len(_drained_escalation_ids) == 1: + _drained_meta["escalation_id"] = _drained_escalation_ids[0] + elif _drained_escalation_ids: + _drained_meta["escalation_ids"] = _drained_escalation_ids # Durable provenance for every `inject` row. `cls` is NOT persisted for this # role (chat_persistence only keeps it for `role == "system"`), and the # frontend's `meta.cronLabel` exists on the wire only because parse_cls_meta diff --git a/src/kiro_crew/dashboard/handlers/__init__.py b/src/kiro_crew/dashboard/handlers/__init__.py index 5221958159f..3e32747ac9d 100644 --- a/src/kiro_crew/dashboard/handlers/__init__.py +++ b/src/kiro_crew/dashboard/handlers/__init__.py @@ -226,6 +226,7 @@ def sel(): # ── Crew Members (handlers/members.py) ── from kiro_crew.dashboard.handlers.members import ( # noqa: E402, F401 api_member_activity, + api_member_conversation, api_member_rules_get, api_member_rules_put, api_member_thread, diff --git a/src/kiro_crew/dashboard/handlers/members.py b/src/kiro_crew/dashboard/handlers/members.py index 612294b78b6..aad770b2fd4 100644 --- a/src/kiro_crew/dashboard/handlers/members.py +++ b/src/kiro_crew/dashboard/handlers/members.py @@ -24,6 +24,7 @@ from aiohttp import web import kiro_crew.dashboard.handlers as _h +from kiro_crew import crew_conversation from kiro_crew import members as members_mod from kiro_crew.config.loader import KiroCrewConfig from kiro_crew.dashboard.chat_persistence import rehydrate_slot_from_history_async @@ -39,6 +40,33 @@ #: not a durability boundary. _ACTIVITY_LIMIT = 50 +#: Per member: the transcript GENERATION the conversation index was last +#: reconciled against (see ``crew_conversation.reconcile_with_transcript``) with +#: nothing deferred — the persisted transcript's mtime plus the live window's +#: length. Recovery is keyed to the transcript, not to the process: a rewind or +#: any other rewrite that removes an escalation row changes the generation, so +#: the index is re-derived on the next read instead of a stale ``needs_you`` +#: surviving for the process lifetime. The re-derivation runs whenever the +#: generation moved, pending or not: a rewind that drops the reply which +#: ANSWERED a record (its card staying) must reopen that record, which is a +#: change with nothing pending beforehand. A member with a deferred +#: (in-flight-grace) record is not recorded, so it is reconciled again on the +#: next roster read. +_RECONCILED: dict[str, tuple[float, int]] = {} + + +def _without_human_reply(row: dict) -> dict: + """A shallow copy of a LIVE (unflushed) transcript row with the human's + reply provenance removed, so recovery reconciliation cannot settle an + escalation on the strength of a reply that is not yet on disk. Rows that + carry no ``human_reply`` are returned as-is; the caller's row is never + mutated (it is the slot's live window).""" + meta = row.get("meta") + if not isinstance(meta, dict) or "human_reply" not in meta: + return row + stripped = {k: v for k, v in meta.items() if k != "human_reply"} + return {**row, "meta": stripped} + def _parse_activity_ts(raw: str) -> float: """Epoch seconds from an activity record's ISO-8601 ``ts``, or 0.0. @@ -219,9 +247,123 @@ def _sanitize(text: str) -> str: row["last_active_ts"] = mt row["last_message"] = preview + # Pending escalations, derived from each member's conversation index — + # one thread hop for the roster, same shape as the binding reads. The + # index is UI state (not the trust binding), so an unreadable file reads + # as "nothing pending" rather than failing the roster. The same hop primes + # the in-memory view the slots projection reads on the event loop, so a + # member whose escalations predate this gateway process gets its badge on + # the first roster load rather than on its next write. + # + # Recovery runs here too: the index is a projection of the transcript, so + # before a member's pending count is trusted the projection is re-derived + # from the transcript (``reconcile_with_transcript``) — orphans (a card row + # the transcript never got) are retracted, durable human replies the live + # hook never recorded are applied. The rows are the PERSISTED transcript's + # (the restored live window is only a tail of it) followed by any live rows + # not yet flushed. A member is marked reconciled for this process only when + # nothing was deferred: a record still inside the in-flight grace is looked + # at again on the next read rather than kept pending forever. There is no + # "nothing pending, skip" short-cut: a rewind can remove the reply that + # ANSWERED a record while its card stays, and reconciliation must reopen + # it — so a transcript change can move the index even with nothing pending. + # The generation check below (a stat, not a read) is what keeps a quiet + # member cheap. + to_reconcile: dict[str, tuple[str, list[dict], int]] = {} + for row in rows: + if state is None or not row["slot_key"]: + continue + live = state._slots.get(row["slot_key"]) + live_rows = list(live.messages) if live else [] + to_reconcile[row["slug"]] = (row["slot_key"], live_rows, len(live_rows)) + + def _read_escalations() -> dict[str, int]: + out: dict[str, int] = {} + # `to_reconcile` is only populated when `state` is set, so the log is + # never None for a job that exists; bound here for the type checker. + log = state.conversation_log if state is not None else None + for row in rows: + try: + crew_conversation.prime(row["slug"]) + job = to_reconcile.get(row["slug"]) + if job is not None and log is not None: + slot_key, live_rows, live_len = job + log_key = f"dashboard:{slot_key}" + generation = (log.session_mtime(log_key) or 0.0, live_len) + if _RECONCILED.get(row["slug"]) == generation and not ( + crew_conversation.index_unreadable(row["slug"]) + ): + job = None # same transcript as last time: nothing to re-derive + if job is not None and log is not None: + persisted = log.read_messages_chained_full(log_key) + seen = { + (m.get("meta") or {}).get("mid") + for m in persisted + if isinstance(m.get("meta"), dict) + } + # Live rows not yet flushed still count for CARD PRESENCE + # (a fresh escalation's row lives only in the window until + # the next flush, and must not read as an orphan) but must + # NOT settle a record: the live hook marks a reply answered + # only after a forced save committed the row, and folding an + # unpersisted reply in here would settle the record on disk + # first — a gateway exit before the flush then restores an + # answered index with no reply in the transcript. Strip the + # human's provenance from the live copies so only persisted + # replies answer; the next flush makes them persisted and the + # generation change re-derives. + transcript = list(persisted) + [ + _without_human_reply(m) + for m in live_rows + if not (isinstance(m.get("meta"), dict) and m["meta"].get("mid") in seen) + ] + outcome = crew_conversation.reconcile_with_transcript( + row["slug"], + transcript, + session_key=slot_key, + member=str(row.get("name") or ""), + ) + if not outcome["deferred"]: + _RECONCILED[row["slug"]] = generation + record = crew_conversation.read_conversation(row["slug"]) + out[row["slug"]] = len(crew_conversation.pending_escalations(record)) + except Exception: # noqa: BLE001 - derived state never fails the roster + out[row["slug"]] = 0 + return out + + pending = await asyncio.to_thread(_read_escalations) + for row in rows: + count = pending.get(row["slug"], 0) if row["slot_key"] else 0 + row["pending_escalations"] = count + row["needs_you"] = count > 0 + return web.json_response({"members": rows}) +async def api_member_conversation(request: web.Request) -> web.Response: + """GET /api/members/{slug}/conversation — the member's thin conversation index. + + Pointers and escalation lifecycle only; bodies live in the referenced + session transcripts. ``needs_you`` and ``pending_escalations`` are derived + on read (a passed deadline reads as ``defaulted``/``expired`` without a + write). Owner-only, like the thread endpoint: the index names sessions. + """ + denied = await _deny_app_caller(request, "members.conversation") + if denied is not None: + return denied + refused = await require_owner_dashboard_request(request, "members.conversation") + if refused is not None: + return refused + try: + slug = members_mod.validate_slug(request.match_info.get("slug", "")) + except members_mod.MemberSlugError: + return web.json_response( + {"error": "invalid member slug", "code": "invalid_member_slug"}, status=400 + ) + record = await asyncio.to_thread(crew_conversation.read_conversation, slug) + return web.json_response(crew_conversation.public_view(record)) + + async def api_member_thread(request: web.Request) -> web.Response: """POST /api/members/{slug}/thread — idempotent get-or-create of a DM thread. @@ -230,8 +372,6 @@ async def api_member_thread(request: web.Request) -> web.Response: re-created (the slot key is a pure derivation of the slug, so re-creation always converges on the same thread). """ - from kiro_crew.dashboard.handlers._shared import require_owner_dashboard_request - denied = await _deny_app_caller(request, "members.thread") if denied is not None: return denied diff --git a/src/kiro_crew/dashboard/handlers/session_control.py b/src/kiro_crew/dashboard/handlers/session_control.py index c904cdc832a..b0d57539338 100644 --- a/src/kiro_crew/dashboard/handlers/session_control.py +++ b/src/kiro_crew/dashboard/handlers/session_control.py @@ -197,6 +197,41 @@ async def api_session_control_send(request: web.Request) -> web.Response: return web.json_response(result) +async def api_session_control_escalate(request: web.Request) -> web.Response: + """POST /api/session-control/escalate — raise something to the human who + owns the caller, as a peer (``session_escalate``). + + Its own route, not a reserved target of ``/send``: the human is not a + session, nothing runs a turn, and a separate route is what lets the caller + gates and the containment policy classify the verb on its own. + """ + refused = await _require_internal(request) + if refused is not None: + return refused + # No prewarm here: `escalate_to_user` warms the config after its own SEL + # prewarm, the same ordering `send_to_target` uses and for the same reason. + state: DashboardState = request.app["state"] + try: + body = await _body(request) + message = body.get("message") + if not isinstance(message, str) or not message.strip(): + raise sc.SessionControlError("message is required", code="message_required") + result = await sc.escalate_to_user( + state, + caller_session_key=_read_session_key(request), + message=message, + # Shapes are validated inside ``escalate_to_user`` — this handler is + # internal-secret gated, so the check lives with the semantics. + deadline=body.get("deadline"), + default_action=body.get("default_action"), + options=body.get("options"), + goal=body.get("goal"), + ) + except sc.SessionControlError as exc: + return _refusal(exc) + return web.json_response(result) + + async def api_session_control_read(request: web.Request) -> web.Response: """GET /api/session-control/read — read another session's transcript tail.""" refused = await _require_internal(request) diff --git a/src/kiro_crew/dashboard/routes/agents.py b/src/kiro_crew/dashboard/routes/agents.py index c62ee0e64cb..60804dfa1eb 100644 --- a/src/kiro_crew/dashboard/routes/agents.py +++ b/src/kiro_crew/dashboard/routes/agents.py @@ -55,3 +55,4 @@ def register(app: web.Application) -> None: app.router.add_get("/api/members/{slug}/activity", handlers.api_member_activity) app.router.add_get("/api/members/{slug}/rules", handlers.api_member_rules_get) app.router.add_put("/api/members/{slug}/rules", handlers.api_member_rules_put) + app.router.add_get("/api/members/{slug}/conversation", handlers.api_member_conversation) diff --git a/src/kiro_crew/dashboard/server.py b/src/kiro_crew/dashboard/server.py index a084f2516fb..84da6687b65 100644 --- a/src/kiro_crew/dashboard/server.py +++ b/src/kiro_crew/dashboard/server.py @@ -445,6 +445,7 @@ async def _should_prevent_sleep(state: DashboardState, port: int) -> bool: "/api/session-control/stop", "/api/session-control/close", "/api/session-control/send", + "/api/session-control/escalate", "/api/session-control/read", # MCP-only structured monitor inspection. The caller selects its # session identity through X-Session-Key, so cookie authentication can @@ -1442,6 +1443,10 @@ def _register_mcp_routes(app: web.Application) -> None: app.router.add_post( "/api/session-control/send", _deferred_session_control("api_session_control_send") ) + app.router.add_post( + "/api/session-control/escalate", + _deferred_session_control("api_session_control_escalate"), + ) app.router.add_get( "/api/session-control/read", _deferred_session_control("api_session_control_read") ) diff --git a/src/kiro_crew/dashboard/session_control.py b/src/kiro_crew/dashboard/session_control.py index adfeec08e63..2ef0073cab5 100644 --- a/src/kiro_crew/dashboard/session_control.py +++ b/src/kiro_crew/dashboard/session_control.py @@ -34,7 +34,9 @@ import asyncio import logging from collections.abc import Callable +from datetime import timezone from typing import TYPE_CHECKING, Any +from urllib.parse import quote from kiro_crew.config.loader import ( KiroCrewConfig, @@ -52,13 +54,20 @@ MAX_SLOTS_PER_CREATOR, SlotOrigin, _safe_folder_tree, + append_and_surface, ) from kiro_crew.dashboard.stop_retry import allow_escalation -from kiro_crew.history import metadata_now_iso, transcript_stem +from kiro_crew.history import metadata_now_iso, mint_row_mid, transcript_stem +from kiro_crew.notifications.bus import NotificationPayload from kiro_crew.security import redact, redact_and_truncate from kiro_crew.sel import sel from kiro_crew.validation import MAX_LONG_STRING +# Deliberately AFTER `validation`: `crew_conversation` imports `members`, which +# imports `artifacts`, which imports `validation` — loading it first re-enters a +# half-initialised `artifacts` (circular import). isort would hoist it; keep it here. +from kiro_crew import crew_conversation as conv # noqa: E402 # isort: skip + if TYPE_CHECKING: # pragma: no cover - typing only from kiro_crew.dashboard.state import DashboardState, _ChatSlot @@ -1940,6 +1949,528 @@ async def send_to_target( return {"ok": True, "target": slot.key, "started": started} +#: Transcript role of an escalation card. A dedicated role (rather than a +#: ``notice`` with a kind) so the chat-profile projection, the Sessions page +#: and the conversation index all key on one word, and so a card can never be +#: mistaken for something the model said. +ESCALATION_ROLE = conv.ESCALATION_ROW_ROLE +ESCALATION_CLS = "msg msg-escalation" + +#: The ``resources`` prefix every escalation SEL denial carries. There is no +#: session target to name (the human is not a slot), so the refusal code is the +#: resource. +_ESCALATION_DENIAL_RESOURCE = "escalate" + +MAX_ESCALATION_OPTION_CHARS = 120 +MAX_ESCALATION_FIELD_CHARS = 500 + + +def _member_slug(caller_key: str) -> str: + """The member slug a DM slot key names, or ``""`` for any other slot.""" + # Lazy for the same layering reason `_member_caller` documents: `members` + # imports `validation`, which sits below this module. + from kiro_crew.members import DM_SLOT_KEY_PREFIX + + if not caller_key.startswith(DM_SLOT_KEY_PREFIX): + return "" + return caller_key[len(DM_SLOT_KEY_PREFIX) :] + + +def _escalation_denied( + caller_session_key: str, reason: str, code: str, status: int = 403 +) -> SessionControlError: + """Build an escalation refusal AND write its SEL denial record. + + Every refusal on the ``session_escalate`` path — caller gates, routing + (``workspace_mismatch`` / owner thread closed), the post-await recheck and + the index write — goes through this one door, so a refusal after + authorization is audited exactly like one before it. + """ + _audit_reason = redact(reason) + _sel_off_loop( + lambda: sel().log_api_access( + caller=f"session:{caller_session_key or 'unknown'}", + operation="session_control.escalate", + outcome="denied", + source="mcp", + resources=f"{_ESCALATION_DENIAL_RESOURCE}:{code}", + error=_audit_reason, + ), + "session-control denial audit", + ) + return SessionControlError(reason, status=status, code=code) + + +def _authorize_escalation_caller( + state: "DashboardState", *, caller_session_key: str +) -> tuple["_ChatSlot", str]: + """The caller-side half of :func:`authorize_target`, for a target that is + not a session. + + The human is not a slot, so none of the target gates apply — but every + CALLER gate does, in the same order and with the same codes, because an + escalation is still a session reaching outside its own transcript: an + unidentified, disabled, unattended, app-scoped, ephemeral, channel-linked or + mirrored caller is refused here exactly as it would be for a peer target. + Denials are SEL-audited under ``session_control.escalate`` like every other + refusal on this surface. Returns the caller's slot and slot key. + """ + + def deny(reason: str, code: str, status: int = 403) -> SessionControlError: + return _escalation_denied(caller_session_key, reason, code, status) + + caller_key = caller_slot_key(state, caller_session_key) + if not caller_key: + raise deny("caller session could not be identified", "caller_unidentified") + if not session_control_enabled() and not _member_caller(caller_key): + raise deny( + "session control is disabled in config (agent.session_control)", + "session_control_disabled", + ) + if caller_key.startswith(UNATTENDED_SLOT_PREFIXES) and not _cron_caller(caller_key): + raise deny( + "unattended sessions (scheduled runs) cannot escalate to the user", + "unattended_caller", + ) + caller_slot = state.get_slot(caller_key) + if caller_slot is None: + raise deny("caller session is no longer open", "caller_gone") + try: + # Same app-scope / app-owned-cron / ephemeral / channel-link / mirror + # gates the create and target paths apply to a caller, from the one + # function that spells them (which evaluates the cron refusal itself, so + # it is not repeated here), so a new caller-side refusal lands on this + # path automatically. + _refuse_ineligible_creator(state, caller_slot) + except SessionControlError as exc: + raise deny(exc.message, exc.code, status=exc.status) from exc + return caller_slot, caller_key + + +def _escalation_home( + state: "DashboardState", caller_slot: "_ChatSlot", caller_key: str, caller_session_key: str +) -> tuple["_ChatSlot", str]: + """Where an escalation from *caller_key* lands, and the owning member's slug. + + * a member DM slot escalates into its own thread — the human reads that + thread, so the card belongs there; + * a worker session a member created (``_created_by`` is a member slot) + escalates into the CREATING member's thread — the human never opened the + worker, the member did; + * any other session escalates into its own transcript (the human is right + there); no member index is touched, so ``needs_you`` stays a member-only + signal. + + The slug is ``""`` when no member owns the conversation. + """ + slug = _member_slug(caller_key) + if slug: + return caller_slot, slug + creator = getattr(caller_slot, "_created_by", "") or "" + creator_slug = _member_slug(creator) if creator else "" + if creator_slug: + member_slot = state.get_slot(creator) + # Same workspace boundary the peer path enforces (`workspace_mismatch`): + # a worker moved to another workspace must not write into its creator's + # thread in the first one. Refused rather than downgraded to "own + # transcript" — the worker asked for the human, and a silent redirect + # would hide that the member never saw it. + if member_slot is not None and getattr(member_slot, "workspace", "default") != getattr( + caller_slot, "workspace", "default" + ): + raise _escalation_denied( + caller_session_key, + "the creating member's thread is in a different workspace", + code="workspace_mismatch", + status=403, + ) + if member_slot is not None: + return member_slot, creator_slug + # The worker HAS an owning member, but the member's thread is not open: + # refuse rather than downgrade to "own transcript". The worker asked for + # the human through its member; a card in a worker nobody is reading, + # with no index record and no badge, would look delivered and be lost. + raise _escalation_denied( + caller_session_key, + "the creating member's thread is not open; escalate again once it is, or " + "report to the member session that owns you", + code="target_gone", + status=409, + ) + return caller_slot, "" + + +def _clean_field(value: Any, *, limit: int, name: str) -> str | None: + if value is None: + return None + if not isinstance(value, str): + raise SessionControlError(f"{name} must be a string", code=f"{name}_invalid", status=400) + text = sanitize_outbound(value.strip()) + if not text: + return None + if len(text) > limit: + raise SessionControlError( + f"{name} exceeds {limit} characters", code=f"{name}_too_long", status=400 + ) + return text + + +def _clean_options(value: Any) -> list[str]: + if value is None: + return [] + if not isinstance(value, list) or not all(isinstance(o, str) for o in value): + raise SessionControlError( + "options must be a list of strings", code="options_invalid", status=400 + ) + cleaned: list[str] = [] + for raw in value: + text = sanitize_outbound(raw.strip()) + if not text: + continue + if len(text) > MAX_ESCALATION_OPTION_CHARS: + raise SessionControlError( + f"an option exceeds {MAX_ESCALATION_OPTION_CHARS} characters", + code="options_too_long", + status=400, + ) + if text not in cleaned: + cleaned.append(text) + if len(cleaned) > conv.MAX_ESCALATION_OPTIONS: + raise SessionControlError( + f"at most {conv.MAX_ESCALATION_OPTIONS} options", code="options_too_many", status=400 + ) + return cleaned + + +async def escalate_to_user( + state: "DashboardState", + *, + caller_session_key: str, + message: str, + deadline: Any = None, + default_action: str | None = None, + options: list[str] | None = None, + goal: str | None = None, +) -> dict[str, Any]: + """``session_escalate``: raise something to the human as a peer. + + Its own tool, not a reserved ``target`` of ``session_send``, because the two + differ in kind: a send delivers text the target session RUNS as a turn; an + escalation writes a transcript row, rings the bell and returns without + starting anything. A separate name is also what lets policy tell them apart + — the channel-agent containment list, the caller-identity gate and the + advertised-set pin each classify ``session_escalate`` on its own. + + Three properties are load-bearing and are what the spec pins: + + * **Non-blocking with a veto window.** Delivery IS success. The caller's + turn continues; nothing waits on the human. When the caller states a + ``deadline`` and a ``default_action``, the card shows both — "unless you + stop me by