diff --git a/.huskies/specs/tech/LLM_CONTEXT_FROM_EVENTS.md b/.huskies/specs/tech/LLM_CONTEXT_FROM_EVENTS.md new file mode 100644 index 00000000..125134ed --- /dev/null +++ b/.huskies/specs/tech/LLM_CONTEXT_FROM_EVENTS.md @@ -0,0 +1,245 @@ +# LLM Context From Events + +Design overview for making any LLM-driven chat persona (Timmy at the +gateway, Sally at a single sled, future personas) aware of huskies +events without the user having to re-narrate them. + +## Goal + +**Update the LLM's context non-intrusively when a state transition +happens.** No new LLM turn is fired; events are simply visible to the +LLM the next time the user (or anything else) does cause it to run. +The LLM should never need to be told what already happened inside +huskies. + +## Guiding Principle + +**Transports have nothing to do with LLMs.** A transport (Matrix bot, +web UI, CLI, future TUIs) is a pure courier — it relays user text in, +LLM text out, and never owns LLM-facing state. Anything the LLM needs +to know lives in huskies, behind a single `assemble_prompt_context` +helper that the transport calls. Adding a new transport must require +zero changes to the event-awareness path. + +## Three things this doc is NOT + +1. **Triggers** — `on StoryMerged{1122} do Rebuild`. These are + deterministic subscribers; they should never invoke the LLM. Covered + in a separate design. +2. **Proactive wake** — running an LLM turn *because* an event fired, + without the user typing. Costs tokens, risks ramble. Explicitly out + of scope here; a separate decision to make later. +3. **A transport feature** — this design assumes any transport that + invokes the LLM uses the same context-assembly helper. Matrix bot, + web UI, CLI all funnel through it. + +## Why Past Attempts Have Failed + +- **Buffer lived on the transport**, not on huskies. The current + `BotContext.pending_pipeline_events` (`server/src/chat/transport/matrix/bot/context.rs:103-116`) + is Matrix-only; web UI users see nothing of the kind, and the buffer + dies with the bot process. +- **Process-local, RAM-only**. Server rebuild → buffer empty. Any + events between the old binary's last user turn and the new binary's + first are silently lost. +- **Unbounded `mpsc` channels drop under lag.** The server logs + routinely show `[xxx-sub] Subscriber lagged, skipped N event(s)`. + When the subscriber feeding the buffer falls behind, events vanish + without being recorded. +- **No end-to-end test.** Nothing asserts "fire event E, send user + message M, the LLM's prompt contains E." +- **No cross-process aggregation.** Events in a sled have no path to + the gateway-side LLM context without bespoke plumbing per event type. + +## Architecture at a Glance + +``` +┌────────────┐ ┌────────────┐ ┌────────────┐ +│ Sled A │ │ Sled B │ │ Sled C │ +│ event_log/ │ │ event_log/ │ │ event_log/ │ ◄── source of truth +└─────┬──────┘ └─────┬──────┘ └─────┬──────┘ (CRDT-backed) + │ │ │ + └───────────────┼───────────────┘ + ▼ + ┌────────────────────┐ + │ Gateway aggregator │ ◄── tail-merges all sled logs + │ event_view/ │ into a single ordered stream + └─────────┬──────────┘ + │ + ▼ + ┌────────────────────────┐ + │ Per-LLM-session state │ ◄── scope filter + + │ sessions// │ high-water mark per stream + └─────────┬──────────────┘ + │ + ▼ + ┌────────────────────────┐ + │ assemble_prompt_context│ ◄── single helper used by + │ (session_id) -> Str │ every transport before + └─────────┬──────────────┘ each LLM turn + │ + ┌──────────────┼──────────────┐ + ▼ ▼ ▼ + Matrix bot Web UI CLI / TUI +``` + +## Event Model — Reuse What Already Exists + +There is no need to invent a parallel event taxonomy. Huskies already +has a complete typed enum and a single broadcast bus: + +- `server/src/pipeline_state/transition.rs` defines `PipelineEvent` + with **30 variants** covering every state-machine transition + (`DepsMet`, `GatesStarted/Passed/Failed`, `QaSkipped`, + `MergeSucceeded/Failed/FailedFinal`, `Accepted`, `Block/Unblock`, + `Abandon`, `Supersede`, `ReviewHold/Cleared`, `Reject`, `Triage`, + `Close`, `Demote`, `Freeze/Unfreeze`, `MergemasterAttempted`, + `FixupRequested`, `ReQueuedForQa`, `MergeAborted`, + `HotfixRequested`, `MergeRetryStarted`). +- The same module defines `ExecutionEvent` with 7 variants for agent + lifecycle (`SpawnRequested`, `SpawnedSuccessfully`, `Heartbeat`, + `HitRateLimit`, `Exited`, `Stopped`, `Reset`). +- Every transition fires a `TransitionFired` event on a single internal + bus. Ten subscribers already consume it (audit-log, + worktree-create-sub, worktree-cleanup-sub, merge-failure-sub, + merge-block-sub, done-archive-sub, content-gc, cost-rollup-sub, + stage-notification-sub, event-triggers). + +**The LLM context injector is just the 11th subscriber on the same +bus.** It writes typed events into the per-sled CRDT event log +described below; everything downstream reuses the existing taxonomy. + +Each persisted entry carries: + +``` +struct LoggedEvent { + id: EventId, // monotonic per sled + sled_id: SledId, + timestamp: UnixSeconds, + transition: TransitionFired, // story_id + from + to + PipelineEvent + // (or ExecutionEvent — see open question) +} +``` + +The few events that genuinely don't fit the pipeline state machine +(e.g. `ProjectAdopted`, `Rebuilt`, `GatewayHealthChanged`) live in a +small, separately-enumerated `InfraEvent` enum, but the same log and +the same subscriber pattern still apply. + +## Session Model + +An LLM session is a first-class CRDT entity: + +``` +struct LlmSession { + id: SessionId, + persona: Persona, // "Timmy", "Sally", ... + scope: ScopeFilter, // { sleds: All } | { sleds: Set } + high_water: BTreeMap, // per-stream + created: UnixSeconds, +} +``` + +The session id is what the transport carries; it's not the Matrix room, +not the web socket id. A given Matrix room may map to one session; a +web UI tab may map to another. Multiple transports for the same human +can share a session if you want — that's a separate UX call. + +## Prompt Assembly Contract + +Every transport calls one helper before invoking the LLM: + +``` +fn assemble_prompt_context(session_id: SessionId) -> String +``` + +Behavior: + +1. Read the session's scope filter and high-water marks. +2. Fetch events from the gateway aggregator that match the scope and + are newer than the high-water marks. +3. Render them as a single `` block, ordered by sled + then timestamp. +4. Advance the high-water marks to the latest event seen, atomically + with the LLM-turn-start CRDT op (so a crash mid-turn doesn't double- + inject). +5. Return the rendered block (empty string if no new events). + +The transport prepends the result to the user's prompt and invokes the +LLM as usual. + +## Persistence & Reliability Rules + +- **Event log is CRDT-backed.** Survives sled restart. +- **High-water marks are CRDT-backed.** Survives gateway restart. +- **Aggregator uses bounded queues with drop-oldest semantics**, and + every drop logs `[event-agg] dropped N events for session ; client + must re-fetch from `. The aggregator never silently + swallows events — if the queue is full, the session gets a sentinel + event `EventStreamGap { from, to }` so the LLM can see it missed + context. +- **End-to-end test required**: `fire(Event::StoryMerged{1122}) → user + sends "what's going on?" → assembled prompt contains "1122 merged"`. + +## Multi-Persona Scoping + +The same machinery serves both Timmy and Sally: + +| Persona | Scope filter | Notes | +|---------|------------------------------------|--------------------------------| +| Timmy | `{ sleds: All }` | Gateway-wide; aware of every sled | +| Sally | `{ sleds: { huskies-server } }` | Single-sled; sled-local events only | +| Manny | `{ sleds: { huskies, ketflix } }` | Hypothetical; subset | + +Sally never has to know Timmy exists, and vice versa. Their sessions +advance their own high-water marks against the same underlying log. + +## Decisions + +| Decision | Choice | Alternative | +|------------------------|-------------------------------------|----------------------------------------------| +| Event publication | Each sled owns its log | Single global log: cross-sled bottleneck | +| Aggregation | Gateway tail-merges | Each session pulls from each sled directly: N×M fanout | +| Buffer location | CRDT-persisted | In-process: lost on rebuild (current bug) | +| Event identity | Typed enum | Strings: structured-log creep, no compile-time safety | +| Drop semantics | Drop-oldest + `EventStreamGap` | Silent drop (current bug): LLM lies confidently | +| Session ↔ transport | Session is separate from transport | One per transport: web tab + Matrix get different views | +| Proactive LLM wake | OUT OF SCOPE | Wake on every event: cost + ramble | + +## Open Questions + +1. **Session lifecycle**. How are sessions created and garbage- + collected? Created on first transport message? GC'd after N days + idle? +2. **Event retention**. How long are events kept in the log? Forever + feels wrong; "since last terminal session turn" feels right but + needs care for multi-session readers. +3. **Multi-transport same session**. Should one human's Matrix and web + UI share a session by default, or always be separate? +4. **Render budget**. If 500 events accumulated between turns, do we + render all 500 or summarize? A `summarize_events` fallback path is + probably worth designing in from the start. +5. **Aggregator placement when there is no gateway**. A standalone + single-sled install has no gateway — does the sled itself host the + aggregator? (Probably yes; trivially "aggregates" its own log.) + +## Phasing + +- **Phase 0 (now):** this design doc. +- **Phase 1:** typed `Event` enum + per-sled CRDT-backed event log; + one publisher subscribes to existing pipeline transitions and writes + `StoryStaged` / `StoryMerged` / `StoryMergeFailed`. +- **Phase 2:** `LlmSession` CRDT entity + `assemble_prompt_context` + helper, wired into the Matrix bot's `handle_message` (replaces the + existing `pending_pipeline_events` Vec). End-to-end test covering the + fire-event → user-turn → prompt-contains-event contract. +- **Phase 3:** Gateway aggregator over multiple sleds; Timmy's session + scoped to `All`. Sally's session scoped to a single sled. +- **Phase 4:** Web UI and any other transports migrated onto + `assemble_prompt_context`; the Matrix-specific Vec deleted. +- **Phase 5:** Bounded queues + `EventStreamGap` sentinel; observability + for `assemble_prompt_context` runs (events injected, gaps observed). + +Each phase ships independently. Phase 2 alone delivers the user-facing +fix: Timmy sees what merged when you next say anything, without you +needing to re-narrate. diff --git a/server/src/chat/transport/discord/commands.rs b/server/src/chat/transport/discord/commands.rs index 621276e0..8b7ccbc5 100644 --- a/server/src/chat/transport/discord/commands.rs +++ b/server/src/chat/transport/discord/commands.rs @@ -362,6 +362,7 @@ async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, use &project_root_str, resume_session_id.as_deref(), None, + None, &mut cancel_rx, move |token| { let mut buf = buffer_for_callback.lock().unwrap(); diff --git a/server/src/chat/transport/matrix/bot/context.rs b/server/src/chat/transport/matrix/bot/context.rs index d62586a0..8f155e12 100644 --- a/server/src/chat/transport/matrix/bot/context.rs +++ b/server/src/chat/transport/matrix/bot/context.rs @@ -109,6 +109,9 @@ pub struct BotContext { /// configured room. Updated atomically on every `on_room_message` call so /// the `health` command can detect a stale or dead sync loop. pub last_matrix_event_ms: Arc, + /// Optional model override from bot.toml. Passed as `--model` to the + /// `claude` CLI when set. + pub model: Option, } impl BotContext { @@ -305,6 +308,7 @@ mod tests { ))), gateway_port: None, last_matrix_event_ms: Arc::new(AtomicI64::new(chrono::Utc::now().timestamp_millis())), + model: None, } } diff --git a/server/src/chat/transport/matrix/bot/messages/handle_message.rs b/server/src/chat/transport/matrix/bot/messages/handle_message.rs index aedf4659..21f4c265 100644 --- a/server/src/chat/transport/matrix/bot/messages/handle_message.rs +++ b/server/src/chat/transport/matrix/bot/messages/handle_message.rs @@ -101,6 +101,7 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message( &project_root_str, resume_session_id.as_deref(), None, + ctx.model.as_deref(), &mut cancel_rx, move |token| { let mut buf = buffer_for_callback.lock().unwrap(); diff --git a/server/src/chat/transport/matrix/bot/run.rs b/server/src/chat/transport/matrix/bot/run.rs index 7377aa36..03bf9081 100644 --- a/server/src/chat/transport/matrix/bot/run.rs +++ b/server/src/chat/transport/matrix/bot/run.rs @@ -337,6 +337,7 @@ pub async fn run_bot( ))), gateway_port, last_matrix_event_ms: Arc::new(AtomicI64::new(chrono::Utc::now().timestamp_millis())), + model: config.model.clone(), }; slog!( diff --git a/server/src/chat/transport/matrix/config/types.rs b/server/src/chat/transport/matrix/config/types.rs index f9c375ae..87430b00 100644 --- a/server/src/chat/transport/matrix/config/types.rs +++ b/server/src/chat/transport/matrix/config/types.rs @@ -71,10 +71,10 @@ pub struct BotConfig { /// (fail-closed). Defaults to 120 seconds. #[serde(default = "default_permission_timeout_secs")] pub permission_timeout_secs: u64, - /// Previously used to select an Anthropic model. Now ignored — the bot - /// uses Claude Code which manages its own model selection. Kept for - /// backwards compatibility so existing bot.toml files still parse. - #[allow(dead_code)] + /// Claude Code model override. When set, passed as `--model ` to + /// the `claude` CLI so the bot uses a specific model instead of the CLI's + /// configured default. + #[serde(default)] pub model: Option, /// Display name the bot uses to identify itself in conversations. /// If unset, the bot falls back to "Assistant". diff --git a/server/src/chat/transport/matrix/health.rs b/server/src/chat/transport/matrix/health.rs index c548d141..e21ebbbf 100644 --- a/server/src/chat/transport/matrix/health.rs +++ b/server/src/chat/transport/matrix/health.rs @@ -694,6 +694,7 @@ mod tests { )), gateway_port: None, last_matrix_event_ms: Arc::new(AtomicI64::new(chrono::Utc::now().timestamp_millis())), + model: None, } } } diff --git a/server/src/chat/transport/matrix/projects.rs b/server/src/chat/transport/matrix/projects.rs index e491119d..79deb8e1 100644 --- a/server/src/chat/transport/matrix/projects.rs +++ b/server/src/chat/transport/matrix/projects.rs @@ -97,6 +97,7 @@ mod tests { )), gateway_port: None, last_matrix_event_ms: Arc::new(AtomicI64::new(chrono::Utc::now().timestamp_millis())), + model: None, }; run_projects_list(&ctx).await } @@ -211,6 +212,7 @@ mod tests { )), gateway_port: None, last_matrix_event_ms: Arc::new(AtomicI64::new(chrono::Utc::now().timestamp_millis())), + model: None, }; let response = run_projects_list(&ctx).await; assert!( diff --git a/server/src/chat/transport/slack/commands/llm.rs b/server/src/chat/transport/slack/commands/llm.rs index d234ed26..9b304589 100644 --- a/server/src/chat/transport/slack/commands/llm.rs +++ b/server/src/chat/transport/slack/commands/llm.rs @@ -66,6 +66,7 @@ pub(super) async fn handle_llm_message( &project_root_str, resume_session_id.as_deref(), None, + None, &mut cancel_rx, move |token| { let mut buf = buffer_for_callback.lock().unwrap(); diff --git a/server/src/chat/transport/whatsapp/commands/llm.rs b/server/src/chat/transport/whatsapp/commands/llm.rs index 5a88c096..c9c80c0a 100644 --- a/server/src/chat/transport/whatsapp/commands/llm.rs +++ b/server/src/chat/transport/whatsapp/commands/llm.rs @@ -65,6 +65,7 @@ pub(super) async fn handle_llm_message( &project_root_str, resume_session_id.as_deref(), None, + None, &mut cancel_rx, move |token| { let mut buf = buffer_for_callback.lock().unwrap(); diff --git a/server/src/http/mcp/dispatch.rs b/server/src/http/mcp/dispatch.rs index 588c4511..88dc0732 100644 --- a/server/src/http/mcp/dispatch.rs +++ b/server/src/http/mcp/dispatch.rs @@ -123,8 +123,8 @@ pub async fn dispatch_tool_call( "git_add" => git_tools::tool_git_add(&args, ctx).await, "git_commit" => git_tools::tool_git_commit(&args, ctx).await, "git_log" => git_tools::tool_git_log(&args, ctx).await, - // Story triage - "status" => status_tools::tool_status(&args, ctx).await, + // Story detail (any stage) + "show" => status_tools::tool_show(&args, ctx).await, // File line count "loc_file" => diagnostics::tool_loc_file(&args, ctx), // Setup wizard tools diff --git a/server/src/http/mcp/status_tools.rs b/server/src/http/mcp/status_tools.rs index b4935170..418c926c 100644 --- a/server/src/http/mcp/status_tools.rs +++ b/server/src/http/mcp/status_tools.rs @@ -150,37 +150,40 @@ async fn git_branch(dir: &Path) -> Option { .flatten() } -pub(super) async fn tool_status(args: &Value, ctx: &AppContext) -> Result { - let story_id = args +pub(super) async fn tool_show(args: &Value, ctx: &AppContext) -> Result { + let raw_id = args .get("story_id") .and_then(|v| v.as_str()) .ok_or("Missing required argument: story_id")?; let root = ctx.state.get_project_root()?; - // Read from CRDT/DB content store — verify the item is in coding. + // Resolve numeric prefix (e.g. "5") to full story_id via find_story_by_number. + let (story_id, contents) = if raw_id.chars().all(|c| c.is_ascii_digit()) { + let (sid, _, _, content) = crate::chat::lookup::find_story_by_number(&root, raw_id) + .ok_or_else(|| { + format!("No work item with number '{raw_id}' found in any pipeline stage.") + })?; + let body = content.ok_or_else(|| { + format!("Work item '{sid}' found in pipeline but its content is unavailable.") + })?; + (sid, body) + } else { + let body = crate::db::read_content(crate::db::ContentKey::Story(raw_id)) + .ok_or_else(|| format!("Work item '{raw_id}' not found in any pipeline stage."))?; + (raw_id.to_string(), body) + }; + + let story_id = story_id.as_str(); + let typed_item = crate::pipeline_state::read_typed(story_id) - .map_err(|e| format!("Failed to read pipeline state: {e}"))? - .ok_or_else(|| format!( - "Story '{story_id}' not found in coding stage. Check the story_id and ensure it is in the current stage." - ))?; - - if !matches!( - typed_item.stage, - crate::pipeline_state::Stage::Coding { .. } - ) { - return Err(format!( - "Story '{story_id}' not found in coding stage. Check the story_id and ensure it is in the current stage." - )); - } - - let contents = crate::db::read_content(crate::db::ContentKey::Story(story_id)) - .ok_or_else(|| format!("Story '{story_id}' has no content in the content store."))?; + .map_err(|e| format!("Failed to read pipeline state: {e}"))?; // --- Metadata (story 929: CRDT-first, yaml_residue marks gaps) --- let mut front_matter = serde_json::Map::new(); if let Some(view) = crate::crdt_state::read_item(story_id) { front_matter.insert("name".to_string(), json!(view.name())); + front_matter.insert("stage".to_string(), json!(view.stage().dir_name())); if let Some(agent) = view.agent() { front_matter.insert("agent".to_string(), json!(agent)); } @@ -195,14 +198,13 @@ pub(super) async fn tool_status(args: &Value, ctx: &AppContext) -> Result claim.as_ref(), crate::pipeline_state::Stage::Merge { claim, .. } => claim.as_ref(), _ => None, - }; + }); if let Some(claim) = stage_claim { front_matter.insert("claimed_by".to_string(), json!(claim.agent.0.as_str())); front_matter.insert( @@ -212,7 +214,6 @@ pub(super) async fn tool_status(args: &Value, ctx: &AppContext) -> Result Vec { }), json!({ "name": "get_pipeline_status", - "description": "Return a structured snapshot of the full work item pipeline. Each item includes only slim fields: story_id, name (capped at 120 chars), stage, agent (with agent_name/model/status), and optional boolean flags blocked and retry_count. Active stages (current, qa, merge, done) appear in the 'active' array; backlog items in 'backlog'. For full story details, use status(story_id) or dump_crdt.", + "description": "Return a structured snapshot of the full work item pipeline. Each item includes only slim fields: story_id, name (capped at 120 chars), stage, agent (with agent_name/model/status), and optional boolean flags blocked and retry_count. Active stages (current, qa, merge, done) appear in the 'active' array; backlog items in 'backlog'. For full story details, use show(story_id).", "inputSchema": { "type": "object", "properties": {} @@ -719,14 +719,14 @@ pub(super) fn story_tools() -> Vec { } }), json!({ - "name": "status", - "description": "Get a full triage dump for an in-progress story: front matter, AC checklist, active worktree/branch, git diff --stat since master, last 5 commits, and last 20 lines of the most recent agent log. Returns a clear error if the story is not in work/2_current/.", + "name": "show", + "description": "Show full details for a work item in any pipeline stage: front matter, AC checklist, and (for coding-stage items) active worktree/branch, git diff --stat since master, last 5 commits, and last 20 lines of the most recent agent log. Accepts a full story_id ('42_story_my_feature') or just the numeric prefix ('42').", "inputSchema": { "type": "object", "properties": { "story_id": { "type": "string", - "description": "Story identifier (filename stem, e.g. '42_story_my_feature')" + "description": "Story identifier — full stem (e.g. '42_story_my_feature') or just the numeric prefix (e.g. '42')" } }, "required": ["story_id"] diff --git a/server/src/llm/chat/run.rs b/server/src/llm/chat/run.rs index 778d58e7..f06b592a 100644 --- a/server/src/llm/chat/run.rs +++ b/server/src/llm/chat/run.rs @@ -209,6 +209,7 @@ where &project_root.to_string_lossy(), config.session_id.as_deref(), None, + None, &mut cancel_rx, |token| on_token(token), |thinking| on_thinking(thinking), diff --git a/server/src/llm/providers/claude_code/mod.rs b/server/src/llm/providers/claude_code/mod.rs index e3af11d9..fc9376d9 100644 --- a/server/src/llm/providers/claude_code/mod.rs +++ b/server/src/llm/providers/claude_code/mod.rs @@ -52,6 +52,7 @@ impl ClaudeCodeProvider { project_root: &str, session_id: Option<&str>, system_prompt: Option<&str>, + model: Option<&str>, cancel_rx: &mut watch::Receiver, mut on_token: F, mut on_thinking: T, @@ -81,6 +82,7 @@ impl ClaudeCodeProvider { let cwd = project_root.to_string(); let resume_id = session_id.map(|s| s.to_string()); let sys_prompt = system_prompt.map(|s| s.to_string()); + let model_override = model.map(|s| s.to_string()); let cancelled_inner = cancelled.clone(); let auth_failed = Arc::new(AtomicBool::new(false)); let auth_failed_clone = auth_failed.clone(); @@ -97,6 +99,7 @@ impl ClaudeCodeProvider { &cwd, resume_id.as_deref(), sys_prompt.as_deref(), + model_override.as_deref(), cancelled_inner, auth_failed_clone, token_tx, @@ -188,6 +191,7 @@ fn run_pty_session( cwd: &str, resume_session_id: Option<&str>, _system_prompt: Option<&str>, + model: Option<&str>, cancelled: Arc, auth_failed: Arc, token_tx: tokio::sync::mpsc::UnboundedSender, @@ -214,6 +218,10 @@ fn run_pty_session( cmd.arg("--resume"); cmd.arg(sid); } + if let Some(m) = model { + cmd.arg("--model"); + cmd.arg(m); + } cmd.arg("--output-format"); cmd.arg("stream-json"); cmd.arg("--verbose"); @@ -236,11 +244,12 @@ fn run_pty_session( cmd.env("CLAUDECODE", ""); slog!( - "[pty-debug] Spawning: claude -p \"{}\" {} --output-format stream-json --verbose --include-partial-messages --permission-prompt-tool mcp__huskies__prompt_permission", + "[pty-debug] Spawning: claude -p \"{}\" {} {} --output-format stream-json --verbose --include-partial-messages --permission-prompt-tool mcp__huskies__prompt_permission", user_message, resume_session_id .map(|s| format!("--resume {s}")) - .unwrap_or_default() + .unwrap_or_default(), + model.map(|m| format!("--model {m}")).unwrap_or_default() ); let mut child = pair