From 4e85238f01ede4e7d3d90515b7eb67ce0477157d Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 01 Jul 2026 14:34:08 +0000 Subject: [PATCH] split agent turn runtime --- .beads/interactions.jsonl | 1 + .beads/issues.jsonl | 1 + klbr-core/src/agent.rs | 732 +++++++----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- klbr-core/src/agent/turn.rs | 596 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ klbr-core/src/tools/local_send.rs | 21 +++++++++++++++++++++ klbr-core/src/tools/mod.rs | 7 ++++++- klbr-core/src/tools/restart_harness.rs | 34 ++++++++++++++++++++++++++++++++++ klbr-core/src/tools/wait_and_continue.rs | 98 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++- 8 file(s) changed, 763 insertion(s)(+), 727 deletion(s)(-) diff --git a/.beads/interactions.jsonl b/.beads/interactions.jsonl --- a/.beads/interactions.jsonl +++ b/.beads/interactions.jsonl @@ -97,3 +97,4 @@ {"id":"int-5dcd101a","kind":"field_change","created_at":"2026-07-01T09:26:28.189416219Z","actor":"dawn","issue_id":"klbr-ds9","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Satisfied by e61b0e5 and 453b014: deduplicated run-loop failure handling and split run_turn into stream/tool phase helpers while preserving behavior; workspace tests passed."}} {"id":"int-706d5ee3","kind":"field_change","created_at":"2026-07-01T09:35:51.422527514Z","actor":"dawn","issue_id":"klbr-6m2","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Split App.svelte into ConnectionRail, ChatHeader, Composer, AddDaemonModal, ReflinkPopover, ResolutionTraces, and shared UI types; verified svelte-check, build, diff check, and local Vite smoke."}} {"id":"int-6a5bb5cc","kind":"field_change","created_at":"2026-07-01T09:48:37.465114937Z","actor":"dawn","issue_id":"klbr-wj2","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Added real folgezettel usage trace report command with reuse, revision, and review flags."}} +{"id":"int-f26ee4d7","kind":"field_change","created_at":"2026-07-01T14:33:43.153698998Z","actor":"dawn","issue_id":"klbr-pic","extra":{"field":"status","new_value":"closed","old_value":"in_progress","reason":"Split turn streaming/tool execution into agent/turn.rs and moved local_send, wait_and_continue, and restart runtime behavior into tool modules while preserving workspace tests."}} diff --git a/.beads/issues.jsonl b/.beads/issues.jsonl --- a/.beads/issues.jsonl +++ b/.beads/issues.jsonl @@ -41,6 +41,7 @@ {"_type":"issue","id":"klbr-wmz.2","title":"Route runtime passive recall through the canonical memory pipeline","description":"Bench retrieval now goes through MemoryPipeline, but runtime passive recall in klbr-core/src/agent.rs still embeds the prompt, calls MemoryStore::get_searchable, runs retrieval::retrieve_exact over legacy memories, then injects Context memory packets. That bypasses fts, exact refs, markdown notes, graph expansion, and the canonical lane/lifecycle path the docs describe.","design":"Avoid duplicating retrieval logic in agent.rs. Either make MemoryPipeline usable by AgentRuntime or extract a shared retrieval facade that both MemoryPipeline and runtime passive recall call.","acceptance_criteria":"Runtime passive recall uses the same lane-aware canonical retrieval and context packet assembly policy as the production pipeline; recalled packets can include fts/exact/dense/graph candidates from refs and markdown notes; archived/tombstoned/suppressed refs do not leak; klbr-core/src/instructions.md matches the actual memory packet format; tests or a focused integration fixture cover passive recall from a markdown note and from an explicit ref.","notes":"Runtime passive recall now calls MemoryPipeline::retrieve_evidence and injects shared EvidencePacket XML via Context::inject_evidence_packets; no-model runtime packet fixture covers turn-window expansion. Remaining acceptance is blocked on klbr-wmz.1 because dense search still starts from legacy memory rows before ref mapping.","status":"closed","priority":1,"issue_type":"feature","assignee":"dawn","owner":"90008@klbr.net","created_at":"2026-06-26T17:53:19Z","created_by":"dawn","updated_at":"2026-06-26T19:44:55Z","started_at":"2026-06-26T19:37:32Z","closed_at":"2026-06-26T19:44:55Z","close_reason":"Completed: runtime passive recall now uses MemoryPipeline/EvidencePlanner packets, instructions document current packet XML, and fixtures cover turn-window recall plus explicit markdown refs; dense canonical dependency completed in klbr-wmz.1.","labels":["architecture","memory","retrieval","runtime"],"dependencies":[{"issue_id":"klbr-wmz.2","depends_on_id":"klbr-wmz","type":"parent-child","created_at":"2026-06-26T20:53:19Z","created_by":"dawn","metadata":"{}"},{"issue_id":"klbr-wmz.2","depends_on_id":"klbr-wmz.1","type":"blocks","created_at":"2026-06-26T22:37:53Z","created_by":"dawn","metadata":"{}"}],"dependency_count":1,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"klbr-wmz.1","title":"Move dense retrieval onto canonical refs and embedding_items","description":"Current source still has dense retrieval on the legacy memory-row surface: klbr-core/src/pipeline.rs search_dense calls MemoryStore::get_searchable and retrieval::retrieve_exact, while fts/exact retrieval uses canonical refs and promptable_text. The schema already has refs, promptable_text, markdown_note_chunks, and embedding_items, so dense retrieval should not be the odd path out.","design":"Prefer a ref-native embedding index backed by embedding_items. Backfill embeddings from promptable_text, keep memory-id aliases as compatibility aliases, and make klbr/full versus dense-only profiles exercise the same canonical identity layer as fts and exact retrieval.","acceptance_criteria":"Dense candidate generation works over canonical ref ids for memories, turn chunks, markdown note chunks, episode notes, profile notes, and procedural notes; lane and lifecycle filtering come from refs/ref_metadata instead of memory tags alone; benchmark traces return canonical refs for dense hits; regression tests cover a markdown-note-only hit and a tombstoned/suppressed ref not leaking through dense search.","status":"closed","priority":1,"issue_type":"feature","assignee":"dawn","owner":"90008@klbr.net","created_at":"2026-06-26T17:53:11Z","created_by":"dawn","updated_at":"2026-06-26T19:43:27Z","started_at":"2026-06-26T19:38:02Z","closed_at":"2026-06-26T19:43:27Z","close_reason":"Completed: dense candidate generation now lazily backfills embedding_items from active promptable refs, scores canonical ref embeddings directly, and tests markdown-note dense hits plus suppressed-ref filtering.","labels":["architecture","memory","refs","retrieval"],"dependencies":[{"issue_id":"klbr-wmz.1","depends_on_id":"klbr-wmz","type":"parent-child","created_at":"2026-06-26T20:53:11Z","created_by":"dawn","metadata":"{}"}],"dependency_count":0,"dependent_count":1,"comment_count":0} {"_type":"issue","id":"klbr-wmz","title":"Finish memory architecture follow-through","description":"Tracks the remaining memory architecture work identified from docs/memory-arch.md, docs/memory-benches.md, docs/memory-implementation-status.md, and current klbr-core/klbr-bench source. Current status says pipeline, typed memory packets, markdown notes, edge mirroring, lifecycle projection, and benchmark runner integration exist; this epic is for gaps still present in source/docs.","acceptance_criteria":"Close when the child issues are complete, docs/memory-implementation-status.md is updated from current verification, and the architecture docs no longer point at missing or stale follow-up work.","status":"closed","priority":1,"issue_type":"epic","owner":"90008@klbr.net","created_at":"2026-06-26T17:52:54Z","created_by":"dawn","updated_at":"2026-06-26T23:45:35Z","closed_at":"2026-06-26T23:45:35Z","close_reason":"All 19 memory architecture child issues are closed; docs/status were updated from current verification; remaining official evaluator run is tracked separately as external blocked klbr-1yn.","labels":["architecture","memory"],"dependency_count":0,"dependent_count":0,"comment_count":0} +{"_type":"issue","id":"klbr-pic","title":"Split agent turn and tool runtime out of agent.rs","description":"agent.rs is still too large after the previous run_turn decomposition. Move the turn streaming/tool iteration machinery and tool-specific runtime behaviors such as wait_and_continue/local_send/restart handling out of the main Agent file while preserving behavior and tests.","status":"closed","priority":2,"issue_type":"task","assignee":"dawn","owner":"90008@klbr.net","created_at":"2026-07-01T14:27:20Z","created_by":"dawn","updated_at":"2026-07-01T14:33:43Z","started_at":"2026-07-01T14:27:24Z","closed_at":"2026-07-01T14:33:43Z","close_reason":"Split turn streaming/tool execution into agent/turn.rs and moved local_send, wait_and_continue, and restart runtime behavior into tool modules while preserving workspace tests.","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"klbr-u03","title":"bridge: refactor hardcoded /home/mayer paths and make binary path configurable","description":"antigravity-bridge/src/auth.rs contains hardcoded /home/mayer home directories and hardcoded binary path BINARY_PATH. Use dirs::home_dir() and make the binary path configurable via environment variables or configuration.","status":"closed","priority":2,"issue_type":"task","assignee":"dawn","owner":"90008@klbr.net","created_at":"2026-06-30T21:42:26Z","created_by":"dawn","updated_at":"2026-06-30T21:59:42Z","started_at":"2026-06-30T21:58:02Z","closed_at":"2026-06-30T21:59:42Z","close_reason":"Replaced hardcoded /home/mayer paths in antigravity bridge auth with dirs::home_dir()-based helpers, added ANTIGRAVITY_BRIDGE_CONFIG_DIR for token/cache files, and made the agy binary path configurable via ANTIGRAVITY_BRIDGE_BINARY_PATH or AGY_BINARY_PATH with ~/.local/bin/agy fallback. Verified cargo fmt --check and cargo test -p antigravity-bridge (9 passed).","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"klbr-noi","title":"nix: fix broken flake.nix referencing non-existent default.nix","description":"flake.nix references ./default.nix to build the default package, but default.nix does not exist in the repository. Refactor flake.nix to output the derivation directly using nix-cargo-integration outputs.","status":"closed","priority":2,"issue_type":"task","assignee":"dawn","owner":"90008@klbr.net","created_at":"2026-06-30T21:42:19Z","created_by":"dawn","updated_at":"2026-06-30T22:01:41Z","started_at":"2026-06-30T22:00:06Z","closed_at":"2026-06-30T22:01:41Z","close_reason":"Replaced the missing ./default.nix package with config.nci.outputs.\"klbr-daemon\".packages.release. Verified nix eval .#packages.x86_64-linux.default.name returns klbr-daemon and nix flake show succeeds without the default.nix error.","dependency_count":0,"dependent_count":0,"comment_count":0} {"_type":"issue","id":"klbr-x1i","title":"bench: replace manual CLI argument parsing with clap","description":"klbr-bench main.rs parses command-line arguments using raw std::env::args and index offsets. This is fragile and lacks auto-help or completions. Port CLI arguments parsing to clap.","status":"closed","priority":2,"issue_type":"task","assignee":"dawn","owner":"90008@klbr.net","created_at":"2026-06-30T21:42:13Z","created_by":"dawn","updated_at":"2026-06-30T22:20:52Z","started_at":"2026-06-30T22:09:02Z","closed_at":"2026-06-30T22:20:52Z","close_reason":"Replaced klbr-bench raw argv dispatch with clap-derived commands and structured runner options","dependency_count":0,"dependent_count":0,"comment_count":0} diff --git a/klbr-core/src/agent.rs b/klbr-core/src/agent.rs --- a/klbr-core/src/agent.rs +++ b/klbr-core/src/agent.rs @@ -12,28 +12,18 @@ harness_block::{self, HarnessBlock, DEFAULT_FORMAT_STYLE}, interrupt::Interrupt, memory::MemoryStore, - models::{LlmClient, LlmEvent, Message, Role, ToolCall}, + models::{LlmClient, LlmEvent, Message, Role}, pipeline::{BenchQuery, ContextBudget, MemoryPipeline}, router::{RouteDecision, Router}, tools::{self, ToolContext}, - AgentEvent, AgentMetrics, CompactionRecord, + AgentEvent, CompactionRecord, }; +use crate::tools::format_duration_human; + +mod turn; + const MAX_CONSECUTIVE_STREAM_FAILURES: usize = 5; - -struct StreamIteration { - response: String, - thinking: String, - tool_calls: Vec, - stream_error: Option, - cancelled_by: Option, -} - -enum ToolIterationOutcome { - Continue, - Suspend(Option), - Restart, -} pub struct Agent { config: Config, @@ -554,611 +544,6 @@ ) .await } - - async fn stream_iteration( - &mut self, - llm: &LlmClient, - ctx: &mut Context, - stashed_interrupt: &mut Option, - ) -> StreamIteration { - let (tok_tx, mut tok_rx) = mpsc::channel(256); - let llm2 = llm.clone(); - let msgs = ctx.as_messages_with_refs(&self.memory); - let defs = self.registry.definitions(); - let stream_task = tokio::spawn(async move { llm2.stream(&msgs, &defs, tok_tx).await }); - - let mut response = String::new(); - let mut thinking = String::new(); - let mut tool_calls = vec![]; - let mut cancelled_by = None; - - loop { - tokio::select! { - biased; - interrupt = self.rx.recv() => { - match interrupt { - Some(int) if int.is_preemptive() => { - tracing::info!(source = int.source_tag(), "preemptive interrupt during stream; aborting"); - stream_task.abort(); - cancelled_by = Some(int); - break; - } - Some(int) => { - // non-preemptive (e.g. ExternalEvent) — stash for after turn - if stashed_interrupt.is_none() { - *stashed_interrupt = Some(int); - } - } - None => {} // channel closed = shutdown - } - } - ev = tok_rx.recv() => { - match ev { - None => break, - Some(LlmEvent::ThinkToken(tok)) => { - thinking.push_str(&tok); - let _ = self.output.send(AgentEvent::ThinkToken(tok)); - } - Some(LlmEvent::Token(tok)) => { - response.push_str(&tok); - let _ = self.output.send(AgentEvent::ScratchToken(tok)); - } - Some(LlmEvent::Usage(usage)) => { - ctx.update_tokens(&usage); - } - Some(LlmEvent::ToolCalls(calls)) => { - tool_calls = calls; - } - Some(LlmEvent::ToolCallStarted) => { - let _ = self.output.send(AgentEvent::ToolCallStarted); - } - } - } - } - } - - let stream_error = match stream_task.await { - Ok(Ok(())) => None, - Ok(Err(e)) => { - let msg = format!("llm stream failed: {e}"); - tracing::error!(%msg); - let _ = self.output.send(AgentEvent::Error(msg.clone())); - let _ = self - .output - .send(AgentEvent::Status("llm stream failed".into())); - Some(msg) - } - Err(e) if e.is_cancelled() => { - // intentional abort — already handled by cancelled_by - None - } - Err(e) => { - let msg = format!("llm stream task panicked/cancelled: {e}"); - tracing::error!(%msg); - let _ = self.output.send(AgentEvent::Error(msg.clone())); - let _ = self - .output - .send(AgentEvent::Status("llm stream task failed".into())); - Some(msg) - } - }; - - StreamIteration { - response, - thinking, - tool_calls, - stream_error, - cancelled_by, - } - } - - async fn execute_tool_iteration( - &mut self, - tool_ctx: &ToolContext, - ctx: &mut Context, - turn_count: &mut usize, - runtime_soul: &mut String, - tool_calls: &[ToolCall], - iteration_thinking: &str, - stashed_interrupt: &mut Option, - consecutive_waits: &mut usize, - discord_send_called: &mut bool, - local_send_called: &mut bool, - ) -> Result { - let reasoning_content = if iteration_thinking.is_empty() { - None - } else { - Some(iteration_thinking.to_string()) - }; - ctx.push_assistant_tool_calls(tool_calls.to_vec(), None, reasoning_content); - let _ = self.memory.log_turn_with_tools( - "assistant", - "", - (!iteration_thinking.is_empty()).then_some(iteration_thinking), - Some(tool_calls), - None, - ); - - let mut pending_local_sends = Vec::new(); - let mut should_restart = false; - for call in tool_calls { - let name = call.function.name.clone(); - if name == "restart_harness" { - should_restart = true; - } - if name != "wait_and_continue" { - *consecutive_waits = 0; - } - if name == "discord_send" { - *discord_send_called = true; - } - if name == "local_send" { - *local_send_called = true; - } - let args = call.function.arguments.clone(); - if name != "local_send" { - let _ = self.output.send(AgentEvent::ToolCall { - name: name.clone(), - args: args.clone(), - }); - } - - let wait_result = if name == "wait_and_continue" { - Some(execute_wait_and_continue(&args, &mut self.rx).await) - } else { - None - }; - let mut incoming_after_tool = None; - let mut suspend_after_wait = false; - let mut sent_local_content = None; - let result = if name == "local_send" { - match tools::local_send_content(&args) { - Ok(content) => { - let result = - format!("sent {} chars to local operator", content.chars().count()); - sent_local_content = Some(content); - result - } - Err(err) => { - tracing::warn!(err = %err, "local_send call had invalid arguments"); - format!("error: {err}") - } - } - } else { - match wait_result { - Some(WaitAndContinueResult::Timeout(mut result)) => { - suspend_after_wait = is_wait_timeout_without_event(&result); - if suspend_after_wait { - *consecutive_waits += 1; - if *consecutive_waits >= 5 && *consecutive_waits % 5 == 0 { - let nudge = format!( - "\n\nnudge: you have called wait_and_continue {} consecutive times without taking other actions. to conserve local resources, please increase the wait duration in your next wait_and_continue call (e.g. wait_and_continue(duration=\"10m\")).", - *consecutive_waits - ); - result.push_str(&nudge); - } - } else { - *consecutive_waits = 0; - } - result - } - Some(WaitAndContinueResult::Incoming { result, interrupt }) => { - incoming_after_tool = Some(interrupt); - result - } - Some(WaitAndContinueResult::Reset) => { - ctx.clear(); - *turn_count = 0; - *runtime_soul = self.sys_prompt()?; - ctx.update_soul(runtime_soul, &[]); - save_context_snapshot(&self.memory, ctx); - let _ = self.output.send(AgentEvent::Reset); - "interrupted by reset; context reset".to_string() - } - Some(WaitAndContinueResult::UpdateSoul(content)) => { - self.memory.set_soul_text(&content)?; - *runtime_soul = self.sys_prompt()?; - let pinned = self.memory.pinned_memories().unwrap_or_default(); - ctx.update_soul(runtime_soul, &pinned); - save_context_snapshot(&self.memory, ctx); - let _ = self.output.send(AgentEvent::Status("soul updated".into())); - "interrupted by soul update; soul updated and continuing".to_string() - } - Some(WaitAndContinueResult::Compact) => { - let _ = self.output.send(AgentEvent::Status("compacting...".into())); - if let Err(e) = compact(tool_ctx, ctx, 0, 0, &self.output).await { - tracing::error!(err = %e, "compaction while waiting failed"); - format!("compaction failed while waiting: {e}") - } else { - *runtime_soul = self.sys_prompt()?; - refresh_context_soul(&self.memory, runtime_soul, ctx); - *turn_count = ctx.turn_count(); - "interrupted by compaction; compacted and continuing".to_string() - } - } - None => { - // Check for a preemptive interrupt before running the tool. - // We use try_recv here (not select!) so the tool future is never - // created and dropped — that would risk double-execution on retry. - match self.rx.try_recv() { - Ok(int) if int.is_preemptive() => { - tracing::info!( - source = int.source_tag(), - tool = name, - "preemptive interrupt before tool execution" - ); - let cancel_result = format!( - "tool execution cancelled by incoming {} message", - int.source_tag() - ); - if name != "local_send" { - let _ = self.output.send(AgentEvent::ToolResult { - name: name.clone(), - content: cancel_result.clone(), - }); - } - ctx.push_tool_result(&call.id, &cancel_result); - let _ = self.memory.log_turn_with_tools( - "tool", - &cancel_result, - None, - None, - Some(&call.id), - ); - flush_pending_local_sends( - &self.memory, - ctx, - turn_count, - &mut pending_local_sends, - ); - return Ok(ToolIterationOutcome::Suspend(Some(int))); - } - Ok(int) => { - // non-preemptive — stash - if stashed_interrupt.is_none() { - *stashed_interrupt = Some(int); - } - } - Err(_) => {} - } - - // Now run the tool, allowing preemptive interrupts during its execution. - let mut tool_output = None; - let mut interrupted_by = None; - let mut exec_fut = Box::pin(self.registry.execute(call, tool_ctx)); - loop { - tokio::select! { - res = &mut exec_fut => { - tool_output = Some(res); - break; - } - int_opt = self.rx.recv() => { - match int_opt { - Some(int) if int.is_preemptive() => { - interrupted_by = Some(int); - break; - } - Some(int) => { - if stashed_interrupt.is_none() { - *stashed_interrupt = Some(int); - } - } - None => {} - } - } - } - } - - if let Some(int) = interrupted_by { - tracing::info!( - source = int.source_tag(), - tool = name, - "tool execution interrupted by incoming preemptive event" - ); - let cancel_result = format!( - "tool execution interrupted by incoming {} message", - int.source_tag() - ); - if name != "local_send" { - let _ = self.output.send(AgentEvent::ToolResult { - name: name.clone(), - content: cancel_result.clone(), - }); - } - ctx.push_tool_result(&call.id, &cancel_result); - let _ = self.memory.log_turn_with_tools( - "tool", - &cancel_result, - None, - None, - Some(&call.id), - ); - flush_pending_local_sends( - &self.memory, - ctx, - turn_count, - &mut pending_local_sends, - ); - return Ok(ToolIterationOutcome::Suspend(Some(int))); - } - - tool_output.unwrap_or_default() - } - } - }; - - if name != "local_send" { - let _ = self.output.send(AgentEvent::ToolResult { - name: name.clone(), - content: result.clone(), - }); - } - - ctx.push_tool_result(&call.id, &result); - let _ = self - .memory - .log_turn_with_tools("tool", &result, None, None, Some(&call.id)); - if let Some(content) = sent_local_content { - let _ = self.output.send(AgentEvent::Token(content.clone())); - pending_local_sends.push(content); - } - if let Some(interrupt) = incoming_after_tool { - flush_pending_local_sends(&self.memory, ctx, turn_count, &mut pending_local_sends); - return Ok(ToolIterationOutcome::Suspend(Some(interrupt))); - } - if suspend_after_wait { - flush_pending_local_sends(&self.memory, ctx, turn_count, &mut pending_local_sends); - return Ok(ToolIterationOutcome::Suspend(None)); - } - // Check for interrupts that arrived *during* tool execution. - // Preemptive ones cut the turn short; non-preemptive are stashed. - match self.rx.try_recv() { - Ok(int) if int.is_preemptive() => { - flush_pending_local_sends( - &self.memory, - ctx, - turn_count, - &mut pending_local_sends, - ); - return Ok(ToolIterationOutcome::Suspend(Some(int))); - } - Ok(int) => { - if stashed_interrupt.is_none() { - *stashed_interrupt = Some(int); - } - } - Err(_) => {} - } - } - - flush_pending_local_sends(&self.memory, ctx, turn_count, &mut pending_local_sends); - if should_restart { - Ok(ToolIterationOutcome::Restart) - } else { - Ok(ToolIterationOutcome::Continue) - } - } - - async fn run_turn( - &mut self, - llm: &LlmClient, - tool_ctx: &ToolContext, - ctx: &mut Context, - turn_count: &mut usize, - runtime_soul: &mut String, - recalled_index: Option, - is_local_message: bool, - is_discord_event: bool, - consecutive_waits: &mut usize, - ) -> Result> { - let mut tool_iterations = 0usize; - let mut discord_send_called = false; - let mut discord_nudge_injected = false; - let mut local_send_called = false; - let mut local_send_nudge_injected = false; - // non-preemptive interrupt received during streaming/tool execution; returned at end of turn - let mut stashed_interrupt: Option = None; - const MAX_TOOL_ITERATIONS: usize = 20; - - // emit Started once — all tokens/thinking for the whole turn go into one bubble - let _ = self.output.send(AgentEvent::Started); - - loop { - let StreamIteration { - response: iteration_response, - thinking: iteration_thinking, - tool_calls, - stream_error, - cancelled_by: stream_cancelled_by, - } = self - .stream_iteration(llm, ctx, &mut stashed_interrupt) - .await; - - if let Some(interrupt) = stream_cancelled_by { - // turn was preempted mid-stream — finish cleanly and hand off - let _ = self.output.send(AgentEvent::Done); - cleanup_recalled_context(ctx, recalled_index); - save_context_snapshot(&self.memory, ctx); - return Ok(Some(interrupt)); - } - - if !tool_calls.is_empty() && tool_iterations < MAX_TOOL_ITERATIONS { - tool_iterations += 1; - let outcome = self - .execute_tool_iteration( - tool_ctx, - ctx, - turn_count, - runtime_soul, - &tool_calls, - &iteration_thinking, - &mut stashed_interrupt, - consecutive_waits, - &mut discord_send_called, - &mut local_send_called, - ) - .await?; - match outcome { - ToolIterationOutcome::Continue => {} - ToolIterationOutcome::Suspend(interrupt) => { - self.finish_suspended_turn(ctx, turn_count, recalled_index) - .await; - return Ok(interrupt); - } - ToolIterationOutcome::Restart => { - self.finish_suspended_turn(ctx, turn_count, recalled_index) - .await; - tracing::info!("restart harness requested, performing restart"); - tokio::time::sleep(std::time::Duration::from_millis(500)).await; - perform_restart(); - } - } - - *runtime_soul = self.sys_prompt()?; - refresh_context_soul(&self.memory, runtime_soul, ctx); - continue; - } - - if let Some(err_msg) = &stream_error { - if iteration_response.is_empty() && tool_calls.is_empty() { - let error_content = format!("error: {err_msg}"); - ctx.push_assistant(&error_content, None); - let _ = self.memory.log_turn("assistant", &error_content, None); - cleanup_recalled_context(ctx, recalled_index); - save_context_snapshot(&self.memory, ctx); - let _ = self.output.send(AgentEvent::Done); - break; - } - } - - // plain assistant text is scratchpad, not delivered speech. - let has_final_text = !iteration_response.is_empty(); - if has_final_text || !iteration_thinking.is_empty() { - ctx.push_assistant( - &scratchpad_context(&iteration_response), - (!iteration_thinking.is_empty()).then_some(iteration_thinking.as_str()), - ); - } - - if has_final_text - && is_local_message - && !local_send_called - && !local_send_nudge_injected - { - local_send_nudge_injected = true; - push_visible_nudge( - ctx, - &self.output, - &plain_text_local_nudge_message(&iteration_response), - ); - tracing::warn!("local_send nudge: assistant emitted plain text without local_send"); - continue; - } - - if has_final_text && is_discord_event && !discord_send_called && !discord_nudge_injected - { - discord_nudge_injected = true; - push_visible_nudge( - ctx, - &self.output, - &plain_text_discord_nudge_message(&iteration_response), - ); - tracing::warn!( - "discord_send nudge: assistant emitted plain text without discord_send" - ); - continue; - } - - self.finish_suspended_turn(ctx, turn_count, recalled_index) - .await; - - if ctx.total_tokens > self.config.watermark_tokens { - let _ = self.output.send(AgentEvent::Status("compacting...".into())); - if let Err(e) = compact( - tool_ctx, - ctx, - self.config.compaction_keep, - self.config.watermark_tokens / 4, - &self.output, - ) - .await - { - tracing::error!(err = %e, "automatic compaction failed"); - let _ = self - .output - .send(AgentEvent::Status(format!("compaction failed: {e}"))); - } else { - *runtime_soul = self.sys_prompt()?; - refresh_context_soul(&self.memory, runtime_soul, ctx); - *turn_count = ctx.turn_count(); - } - } - - *consecutive_waits = 0; - break; - } - - Ok(stashed_interrupt) - } - - async fn finish_suspended_turn( - &self, - ctx: &mut Context, - turn_count: &usize, - recalled_index: Option, - ) { - cleanup_recalled_context(ctx, recalled_index); - save_context_snapshot(&self.memory, ctx); - - let _ = self.output.send(AgentEvent::Done); - let metrics = AgentMetrics { - turn_count: *turn_count, - context_tokens: ctx.total_tokens, - watermark: self.config.watermark_tokens, - prompt_tokens: ctx.prompt_tokens, - completion_tokens: ctx.completion_tokens, - cached_prompt_tokens: ctx.cached_prompt_tokens, - prompt_processing_ms: ctx.prompt_processing_ms, - generation_ms: ctx.generation_ms, - bridge_processing_ms: ctx.bridge_processing_ms, - }; - *self.snapshot.write().await = Some(metrics.clone()); - let _ = self.output.send(AgentEvent::Metrics(metrics)); - } -} - -#[cfg(unix)] -fn perform_restart() -> ! { - let exe = match std::env::current_exe() { - Ok(path) => path, - Err(e) => { - tracing::error!("failed to get current executable path: {e}"); - std::process::exit(1); - } - }; - let args: Vec = std::env::args().skip(1).collect(); - - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or_default(); - - tracing::info!("restarting harness: {:?} with args {:?}", exe, args); - - use std::os::unix::process::CommandExt; - let mut cmd = std::process::Command::new(exe); - cmd.args(&args); - cmd.env("KLBR_RESTART_TIMESTAMP", now.to_string()); - - let err = cmd.exec(); - tracing::error!("exec failed: {err}"); - std::process::exit(1); -} - -#[cfg(not(unix))] -fn perform_restart() -> ! { - tracing::warn!("restart only supported on unix systems"); - std::process::exit(0); } fn unix_timestamp() -> i64 { @@ -1296,7 +681,7 @@ let Some(last) = ctx.turns.last() else { return false; }; - if last.role != Role::Tool || !is_wait_timeout_without_event(last.content_str()) { + if last.role != Role::Tool || !tools::wait_timeout_without_event(last.content_str()) { return false; } let Some(tool_call_id) = last.tool_call_id.as_deref() else { @@ -1314,109 +699,6 @@ .iter() .any(|call| call.id == tool_call_id && call.function.name == "wait_and_continue") }) -} - -fn is_wait_timeout_without_event(result: &str) -> bool { - result.contains("no incoming event arrived") -} - -enum WaitAndContinueResult { - Timeout(String), - Incoming { - result: String, - interrupt: Interrupt, - }, - Reset, - Compact, - UpdateSoul(String), -} - -async fn execute_wait_and_continue( - arguments: &str, - rx: &mut mpsc::Receiver, -) -> WaitAndContinueResult { - let args: serde_json::Value = serde_json::from_str(arguments).unwrap_or_default(); - let timeout = match tools::wait_and_continue_timeout(&args) { - Ok(timeout) => timeout, - Err(err) => return WaitAndContinueResult::Timeout(format!("error: {err}")), - }; - - let start_time = std::time::Instant::now(); - let mut sleep = Box::pin(tokio::time::sleep(timeout)); - tokio::select! { - _ = &mut sleep => { - let elapsed = start_time.elapsed(); - let elapsed_str = format_duration_human(elapsed); - let current_time = chrono::Local::now().format("%Y-%m-%d %H:%M:%S %Z").to_string(); - WaitAndContinueResult::Timeout(format!( - "waited {elapsed_str}; no incoming event arrived; current time: {current_time}" - )) - } - interrupt = rx.recv() => { - let elapsed = start_time.elapsed(); - let elapsed_str = format_duration_human(elapsed); - let current_time = chrono::Local::now().format("%Y-%m-%d %H:%M:%S %Z").to_string(); - match interrupt { - Some(Interrupt::Reset) => WaitAndContinueResult::Reset, - Some(Interrupt::Compact) => WaitAndContinueResult::Compact, - Some(Interrupt::UpdateSoul { content }) => WaitAndContinueResult::UpdateSoul(content), - Some(Interrupt::DebugReflectRequest) => WaitAndContinueResult::Timeout( - format!("debug reflection request received while waiting; current time: {current_time}; elapsed since last turn: {elapsed_str}; try again after this turn") - ), - Some(interrupt) => { - let source = interrupt.source_tag().to_string(); - WaitAndContinueResult::Incoming { - result: format!( - "interrupted early by incoming {source} event after {elapsed_str}; current time: {current_time}" - ), - interrupt, - } - } - None => WaitAndContinueResult::Timeout(format!("interrupt channel closed; current time: {current_time}; elapsed since last turn: {elapsed_str}; continue")), - } - } - } -} - -fn format_duration_human(d: std::time::Duration) -> String { - let secs = d.as_secs(); - if secs == 0 { - return format!("{}ms", d.as_millis()); - } - let h = secs / 3600; - let m = (secs % 3600) / 60; - let s = secs % 60; - let mut parts = vec![]; - if h > 0 { - parts.push(format!("{h}hour")); - } - if m > 0 { - parts.push(format!("{m}min")); - } - if s > 0 || parts.is_empty() { - parts.push(format!("{s}sec")); - } - parts.join(" ") -} - -fn inject_incoming_interrupt( - interrupt: Interrupt, - agent_name: &str, - ctx: &mut Context, - memory: &MemoryStore, - output: &broadcast::Sender, - turn_count: &mut usize, -) { - let prompt_text = interrupt.prompt_text(agent_name); - ctx.push_input(&prompt_text); - if let Ok(entry) = memory.log_turn("user", &prompt_text, None) { - let _ = output.send(AgentEvent::UserTurn(entry)); - } - *turn_count += 1; - let _ = output.send(AgentEvent::Status(format!( - "injected incoming {} event", - interrupt.source_tag() - ))); } fn save_context_snapshot(memory: &MemoryStore, ctx: &Context) { diff --git a/klbr-core/src/agent/turn.rs b/klbr-core/src/agent/turn.rs new file mode 100644 --- /dev/null +++ b/klbr-core/src/agent/turn.rs @@ -0,0 +1,596 @@ +use anyhow::Result; +use tokio::sync::mpsc; + +use super::{ + cleanup_recalled_context, compact, flush_pending_local_sends, plain_text_discord_nudge_message, + plain_text_local_nudge_message, push_visible_nudge, refresh_context_soul, + save_context_snapshot, scratchpad_context, Agent, +}; +use crate::{ + context::Context, + interrupt::Interrupt, + models::{LlmClient, LlmEvent, ToolCall}, + tools::{self, RuntimeWaitResult as WaitAndContinueResult, ToolContext}, + AgentEvent, AgentMetrics, +}; + +struct StreamIteration { + response: String, + thinking: String, + tool_calls: Vec, + stream_error: Option, + cancelled_by: Option, +} + +enum ToolIterationOutcome { + Continue, + Suspend(Option), + Restart, +} + +impl Agent { + async fn stream_iteration( + &mut self, + llm: &LlmClient, + ctx: &mut Context, + stashed_interrupt: &mut Option, + ) -> StreamIteration { + let (tok_tx, mut tok_rx) = mpsc::channel(256); + let llm2 = llm.clone(); + let msgs = ctx.as_messages_with_refs(&self.memory); + let defs = self.registry.definitions(); + let stream_task = tokio::spawn(async move { llm2.stream(&msgs, &defs, tok_tx).await }); + + let mut response = String::new(); + let mut thinking = String::new(); + let mut tool_calls = vec![]; + let mut cancelled_by = None; + + loop { + tokio::select! { + biased; + interrupt = self.rx.recv() => { + match interrupt { + Some(int) if int.is_preemptive() => { + tracing::info!(source = int.source_tag(), "preemptive interrupt during stream; aborting"); + stream_task.abort(); + cancelled_by = Some(int); + break; + } + Some(int) => { + // non-preemptive (e.g. ExternalEvent) — stash for after turn + if stashed_interrupt.is_none() { + *stashed_interrupt = Some(int); + } + } + None => {} // channel closed = shutdown + } + } + ev = tok_rx.recv() => { + match ev { + None => break, + Some(LlmEvent::ThinkToken(tok)) => { + thinking.push_str(&tok); + let _ = self.output.send(AgentEvent::ThinkToken(tok)); + } + Some(LlmEvent::Token(tok)) => { + response.push_str(&tok); + let _ = self.output.send(AgentEvent::ScratchToken(tok)); + } + Some(LlmEvent::Usage(usage)) => { + ctx.update_tokens(&usage); + } + Some(LlmEvent::ToolCalls(calls)) => { + tool_calls = calls; + } + Some(LlmEvent::ToolCallStarted) => { + let _ = self.output.send(AgentEvent::ToolCallStarted); + } + } + } + } + } + + let stream_error = match stream_task.await { + Ok(Ok(())) => None, + Ok(Err(e)) => { + let msg = format!("llm stream failed: {e}"); + tracing::error!(%msg); + let _ = self.output.send(AgentEvent::Error(msg.clone())); + let _ = self + .output + .send(AgentEvent::Status("llm stream failed".into())); + Some(msg) + } + Err(e) if e.is_cancelled() => { + // intentional abort — already handled by cancelled_by + None + } + Err(e) => { + let msg = format!("llm stream task panicked/cancelled: {e}"); + tracing::error!(%msg); + let _ = self.output.send(AgentEvent::Error(msg.clone())); + let _ = self + .output + .send(AgentEvent::Status("llm stream task failed".into())); + Some(msg) + } + }; + + StreamIteration { + response, + thinking, + tool_calls, + stream_error, + cancelled_by, + } + } + + async fn execute_tool_iteration( + &mut self, + tool_ctx: &ToolContext, + ctx: &mut Context, + turn_count: &mut usize, + runtime_soul: &mut String, + tool_calls: &[ToolCall], + iteration_thinking: &str, + stashed_interrupt: &mut Option, + consecutive_waits: &mut usize, + discord_send_called: &mut bool, + local_send_called: &mut bool, + ) -> Result { + let reasoning_content = if iteration_thinking.is_empty() { + None + } else { + Some(iteration_thinking.to_string()) + }; + ctx.push_assistant_tool_calls(tool_calls.to_vec(), None, reasoning_content); + let _ = self.memory.log_turn_with_tools( + "assistant", + "", + (!iteration_thinking.is_empty()).then_some(iteration_thinking), + Some(tool_calls), + None, + ); + + let mut pending_local_sends = Vec::new(); + let mut should_restart = false; + for call in tool_calls { + let name = call.function.name.clone(); + if name == "restart_harness" { + should_restart = true; + } + if name != "wait_and_continue" { + *consecutive_waits = 0; + } + if name == "discord_send" { + *discord_send_called = true; + } + if name == "local_send" { + *local_send_called = true; + } + let args = call.function.arguments.clone(); + if name != "local_send" { + let _ = self.output.send(AgentEvent::ToolCall { + name: name.clone(), + args: args.clone(), + }); + } + + let wait_result = if name == "wait_and_continue" { + Some(tools::execute_wait_and_continue(&args, &mut self.rx).await) + } else { + None + }; + let mut incoming_after_tool = None; + let mut suspend_after_wait = false; + let mut sent_local_content = None; + let result = if name == "local_send" { + let delivery = tools::local_send_runtime_delivery(&args); + if let Some(err) = &delivery.error { + tracing::warn!(err = %err, "local_send call had invalid arguments"); + } + sent_local_content = delivery.sent_content; + delivery.result + } else { + match wait_result { + Some(WaitAndContinueResult::Timeout(mut result)) => { + suspend_after_wait = tools::wait_timeout_without_event(&result); + if suspend_after_wait { + *consecutive_waits += 1; + if *consecutive_waits >= 5 && *consecutive_waits % 5 == 0 { + let nudge = format!( + "\n\nnudge: you have called wait_and_continue {} consecutive times without taking other actions. to conserve local resources, please increase the wait duration in your next wait_and_continue call (e.g. wait_and_continue(duration=\"10m\")).", + *consecutive_waits + ); + result.push_str(&nudge); + } + } else { + *consecutive_waits = 0; + } + result + } + Some(WaitAndContinueResult::Incoming { result, interrupt }) => { + incoming_after_tool = Some(interrupt); + result + } + Some(WaitAndContinueResult::Reset) => { + ctx.clear(); + *turn_count = 0; + *runtime_soul = self.sys_prompt()?; + ctx.update_soul(runtime_soul, &[]); + save_context_snapshot(&self.memory, ctx); + let _ = self.output.send(AgentEvent::Reset); + "interrupted by reset; context reset".to_string() + } + Some(WaitAndContinueResult::UpdateSoul(content)) => { + self.memory.set_soul_text(&content)?; + *runtime_soul = self.sys_prompt()?; + let pinned = self.memory.pinned_memories().unwrap_or_default(); + ctx.update_soul(runtime_soul, &pinned); + save_context_snapshot(&self.memory, ctx); + let _ = self.output.send(AgentEvent::Status("soul updated".into())); + "interrupted by soul update; soul updated and continuing".to_string() + } + Some(WaitAndContinueResult::Compact) => { + let _ = self.output.send(AgentEvent::Status("compacting...".into())); + if let Err(e) = compact(tool_ctx, ctx, 0, 0, &self.output).await { + tracing::error!(err = %e, "compaction while waiting failed"); + format!("compaction failed while waiting: {e}") + } else { + *runtime_soul = self.sys_prompt()?; + refresh_context_soul(&self.memory, runtime_soul, ctx); + *turn_count = ctx.turn_count(); + "interrupted by compaction; compacted and continuing".to_string() + } + } + None => { + // Check for a preemptive interrupt before running the tool. + // We use try_recv here (not select!) so the tool future is never + // created and dropped — that would risk double-execution on retry. + match self.rx.try_recv() { + Ok(int) if int.is_preemptive() => { + tracing::info!( + source = int.source_tag(), + tool = name, + "preemptive interrupt before tool execution" + ); + let cancel_result = format!( + "tool execution cancelled by incoming {} message", + int.source_tag() + ); + if name != "local_send" { + let _ = self.output.send(AgentEvent::ToolResult { + name: name.clone(), + content: cancel_result.clone(), + }); + } + ctx.push_tool_result(&call.id, &cancel_result); + let _ = self.memory.log_turn_with_tools( + "tool", + &cancel_result, + None, + None, + Some(&call.id), + ); + flush_pending_local_sends( + &self.memory, + ctx, + turn_count, + &mut pending_local_sends, + ); + return Ok(ToolIterationOutcome::Suspend(Some(int))); + } + Ok(int) => { + // non-preemptive — stash + if stashed_interrupt.is_none() { + *stashed_interrupt = Some(int); + } + } + Err(_) => {} + } + + // Now run the tool, allowing preemptive interrupts during its execution. + let mut tool_output = None; + let mut interrupted_by = None; + let mut exec_fut = Box::pin(self.registry.execute(call, tool_ctx)); + loop { + tokio::select! { + res = &mut exec_fut => { + tool_output = Some(res); + break; + } + int_opt = self.rx.recv() => { + match int_opt { + Some(int) if int.is_preemptive() => { + interrupted_by = Some(int); + break; + } + Some(int) => { + if stashed_interrupt.is_none() { + *stashed_interrupt = Some(int); + } + } + None => {} + } + } + } + } + + if let Some(int) = interrupted_by { + tracing::info!( + source = int.source_tag(), + tool = name, + "tool execution interrupted by incoming preemptive event" + ); + let cancel_result = format!( + "tool execution interrupted by incoming {} message", + int.source_tag() + ); + if name != "local_send" { + let _ = self.output.send(AgentEvent::ToolResult { + name: name.clone(), + content: cancel_result.clone(), + }); + } + ctx.push_tool_result(&call.id, &cancel_result); + let _ = self.memory.log_turn_with_tools( + "tool", + &cancel_result, + None, + None, + Some(&call.id), + ); + flush_pending_local_sends( + &self.memory, + ctx, + turn_count, + &mut pending_local_sends, + ); + return Ok(ToolIterationOutcome::Suspend(Some(int))); + } + + tool_output.unwrap_or_default() + } + } + }; + + if name != "local_send" { + let _ = self.output.send(AgentEvent::ToolResult { + name: name.clone(), + content: result.clone(), + }); + } + + ctx.push_tool_result(&call.id, &result); + let _ = self + .memory + .log_turn_with_tools("tool", &result, None, None, Some(&call.id)); + if let Some(content) = sent_local_content { + let _ = self.output.send(AgentEvent::Token(content.clone())); + pending_local_sends.push(content); + } + if let Some(interrupt) = incoming_after_tool { + flush_pending_local_sends(&self.memory, ctx, turn_count, &mut pending_local_sends); + return Ok(ToolIterationOutcome::Suspend(Some(interrupt))); + } + if suspend_after_wait { + flush_pending_local_sends(&self.memory, ctx, turn_count, &mut pending_local_sends); + return Ok(ToolIterationOutcome::Suspend(None)); + } + // Check for interrupts that arrived *during* tool execution. + // Preemptive ones cut the turn short; non-preemptive are stashed. + match self.rx.try_recv() { + Ok(int) if int.is_preemptive() => { + flush_pending_local_sends( + &self.memory, + ctx, + turn_count, + &mut pending_local_sends, + ); + return Ok(ToolIterationOutcome::Suspend(Some(int))); + } + Ok(int) => { + if stashed_interrupt.is_none() { + *stashed_interrupt = Some(int); + } + } + Err(_) => {} + } + } + + flush_pending_local_sends(&self.memory, ctx, turn_count, &mut pending_local_sends); + if should_restart { + Ok(ToolIterationOutcome::Restart) + } else { + Ok(ToolIterationOutcome::Continue) + } + } + + pub(super) async fn run_turn( + &mut self, + llm: &LlmClient, + tool_ctx: &ToolContext, + ctx: &mut Context, + turn_count: &mut usize, + runtime_soul: &mut String, + recalled_index: Option, + is_local_message: bool, + is_discord_event: bool, + consecutive_waits: &mut usize, + ) -> Result> { + let mut tool_iterations = 0usize; + let mut discord_send_called = false; + let mut discord_nudge_injected = false; + let mut local_send_called = false; + let mut local_send_nudge_injected = false; + // non-preemptive interrupt received during streaming/tool execution; returned at end of turn + let mut stashed_interrupt: Option = None; + const MAX_TOOL_ITERATIONS: usize = 20; + + // emit Started once — all tokens/thinking for the whole turn go into one bubble + let _ = self.output.send(AgentEvent::Started); + + loop { + let StreamIteration { + response: iteration_response, + thinking: iteration_thinking, + tool_calls, + stream_error, + cancelled_by: stream_cancelled_by, + } = self + .stream_iteration(llm, ctx, &mut stashed_interrupt) + .await; + + if let Some(interrupt) = stream_cancelled_by { + // turn was preempted mid-stream — finish cleanly and hand off + let _ = self.output.send(AgentEvent::Done); + cleanup_recalled_context(ctx, recalled_index); + save_context_snapshot(&self.memory, ctx); + return Ok(Some(interrupt)); + } + + if !tool_calls.is_empty() && tool_iterations < MAX_TOOL_ITERATIONS { + tool_iterations += 1; + let outcome = self + .execute_tool_iteration( + tool_ctx, + ctx, + turn_count, + runtime_soul, + &tool_calls, + &iteration_thinking, + &mut stashed_interrupt, + consecutive_waits, + &mut discord_send_called, + &mut local_send_called, + ) + .await?; + match outcome { + ToolIterationOutcome::Continue => {} + ToolIterationOutcome::Suspend(interrupt) => { + self.finish_suspended_turn(ctx, turn_count, recalled_index) + .await; + return Ok(interrupt); + } + ToolIterationOutcome::Restart => { + self.finish_suspended_turn(ctx, turn_count, recalled_index) + .await; + tracing::info!("restart harness requested, performing restart"); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + tools::perform_restart(); + } + } + + *runtime_soul = self.sys_prompt()?; + refresh_context_soul(&self.memory, runtime_soul, ctx); + continue; + } + + if let Some(err_msg) = &stream_error { + if iteration_response.is_empty() && tool_calls.is_empty() { + let error_content = format!("error: {err_msg}"); + ctx.push_assistant(&error_content, None); + let _ = self.memory.log_turn("assistant", &error_content, None); + cleanup_recalled_context(ctx, recalled_index); + save_context_snapshot(&self.memory, ctx); + let _ = self.output.send(AgentEvent::Done); + break; + } + } + + // plain assistant text is scratchpad, not delivered speech. + let has_final_text = !iteration_response.is_empty(); + if has_final_text || !iteration_thinking.is_empty() { + ctx.push_assistant( + &scratchpad_context(&iteration_response), + (!iteration_thinking.is_empty()).then_some(iteration_thinking.as_str()), + ); + } + + if has_final_text + && is_local_message + && !local_send_called + && !local_send_nudge_injected + { + local_send_nudge_injected = true; + push_visible_nudge( + ctx, + &self.output, + &plain_text_local_nudge_message(&iteration_response), + ); + tracing::warn!("local_send nudge: assistant emitted plain text without local_send"); + continue; + } + + if has_final_text && is_discord_event && !discord_send_called && !discord_nudge_injected + { + discord_nudge_injected = true; + push_visible_nudge( + ctx, + &self.output, + &plain_text_discord_nudge_message(&iteration_response), + ); + tracing::warn!( + "discord_send nudge: assistant emitted plain text without discord_send" + ); + continue; + } + + self.finish_suspended_turn(ctx, turn_count, recalled_index) + .await; + + if ctx.total_tokens > self.config.watermark_tokens { + let _ = self.output.send(AgentEvent::Status("compacting...".into())); + if let Err(e) = compact( + tool_ctx, + ctx, + self.config.compaction_keep, + self.config.watermark_tokens / 4, + &self.output, + ) + .await + { + tracing::error!(err = %e, "automatic compaction failed"); + let _ = self + .output + .send(AgentEvent::Status(format!("compaction failed: {e}"))); + } else { + *runtime_soul = self.sys_prompt()?; + refresh_context_soul(&self.memory, runtime_soul, ctx); + *turn_count = ctx.turn_count(); + } + } + + *consecutive_waits = 0; + break; + } + + Ok(stashed_interrupt) + } + + async fn finish_suspended_turn( + &self, + ctx: &mut Context, + turn_count: &usize, + recalled_index: Option, + ) { + cleanup_recalled_context(ctx, recalled_index); + save_context_snapshot(&self.memory, ctx); + + let _ = self.output.send(AgentEvent::Done); + let metrics = AgentMetrics { + turn_count: *turn_count, + context_tokens: ctx.total_tokens, + watermark: self.config.watermark_tokens, + prompt_tokens: ctx.prompt_tokens, + completion_tokens: ctx.completion_tokens, + cached_prompt_tokens: ctx.cached_prompt_tokens, + prompt_processing_ms: ctx.prompt_processing_ms, + generation_ms: ctx.generation_ms, + bridge_processing_ms: ctx.bridge_processing_ms, + }; + *self.snapshot.write().await = Some(metrics.clone()); + let _ = self.output.send(AgentEvent::Metrics(metrics)); + } +} diff --git a/klbr-core/src/tools/local_send.rs b/klbr-core/src/tools/local_send.rs --- a/klbr-core/src/tools/local_send.rs +++ b/klbr-core/src/tools/local_send.rs @@ -46,6 +46,27 @@ content_from_value(&value) } +pub struct RuntimeDelivery { + pub result: String, + pub sent_content: Option, + pub error: Option, +} + +pub fn runtime_delivery(args: &str) -> RuntimeDelivery { + match content_from_args(args) { + Ok(content) => RuntimeDelivery { + result: format!("sent {} chars to local operator", content.chars().count()), + sent_content: Some(content), + error: None, + }, + Err(err) => RuntimeDelivery { + result: format!("error: {err}"), + sent_content: None, + error: Some(err), + }, + } +} + fn content_from_value(value: &serde_json::Value) -> Result { let content = value["content"] .as_str() diff --git a/klbr-core/src/tools/mod.rs b/klbr-core/src/tools/mod.rs --- a/klbr-core/src/tools/mod.rs +++ b/klbr-core/src/tools/mod.rs @@ -22,7 +22,12 @@ use crate::models::{LlmClient, ToolCall, ToolDef}; pub use local_send::content_from_args as local_send_content; -pub use wait_and_continue::timeout_duration as wait_and_continue_timeout; +pub use local_send::{runtime_delivery as local_send_runtime_delivery, RuntimeDelivery}; +pub use restart_harness::perform as perform_restart; +pub use wait_and_continue::{ + execute_runtime as execute_wait_and_continue, format_duration_human, + timeout_duration as wait_and_continue_timeout, wait_timeout_without_event, RuntimeWaitResult, +}; /// runtime dependencies passed to every tool execution #[derive(Clone)] diff --git a/klbr-core/src/tools/restart_harness.rs b/klbr-core/src/tools/restart_harness.rs --- a/klbr-core/src/tools/restart_harness.rs +++ b/klbr-core/src/tools/restart_harness.rs @@ -28,3 +28,37 @@ ) -> Pin + Send>> { Box::pin(async { "harness restarting... websocket will reconnect.".to_string() }) } + +#[cfg(unix)] +pub fn perform() -> ! { + let exe = match std::env::current_exe() { + Ok(path) => path, + Err(e) => { + tracing::error!("failed to get current executable path: {e}"); + std::process::exit(1); + } + }; + let args: Vec = std::env::args().skip(1).collect(); + + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or_default(); + + tracing::info!("restarting harness: {:?} with args {:?}", exe, args); + + use std::os::unix::process::CommandExt; + let mut cmd = std::process::Command::new(exe); + cmd.args(&args); + cmd.env("KLBR_RESTART_TIMESTAMP", now.to_string()); + + let err = cmd.exec(); + tracing::error!("exec failed: {err}"); + std::process::exit(1); +} + +#[cfg(not(unix))] +pub fn perform() -> ! { + tracing::warn!("restart only supported on unix systems"); + std::process::exit(0); +} diff --git a/klbr-core/src/tools/wait_and_continue.rs b/klbr-core/src/tools/wait_and_continue.rs --- a/klbr-core/src/tools/wait_and_continue.rs +++ b/klbr-core/src/tools/wait_and_continue.rs @@ -3,7 +3,9 @@ use std::time::Duration; use serde_json::json; +use tokio::sync::mpsc; +use crate::interrupt::Interrupt; use crate::models::ToolDef; use super::{Tool, ToolContext}; @@ -81,11 +83,94 @@ Err("duration is required (either 'timeout_ms' or 'seconds' must be specified)".to_string()) } +pub enum RuntimeWaitResult { + Timeout(String), + Incoming { + result: String, + interrupt: Interrupt, + }, + Reset, + Compact, + UpdateSoul(String), +} + +pub async fn execute_runtime( + arguments: &str, + rx: &mut mpsc::Receiver, +) -> RuntimeWaitResult { + let args: serde_json::Value = serde_json::from_str(arguments).unwrap_or_default(); + let timeout = match timeout_duration(&args) { + Ok(timeout) => timeout, + Err(err) => return RuntimeWaitResult::Timeout(format!("error: {err}")), + }; + + let start_time = std::time::Instant::now(); + let mut sleep = Box::pin(tokio::time::sleep(timeout)); + tokio::select! { + _ = &mut sleep => { + let elapsed = start_time.elapsed(); + let elapsed_str = format_duration_human(elapsed); + let current_time = chrono::Local::now().format("%Y-%m-%d %H:%M:%S %Z").to_string(); + RuntimeWaitResult::Timeout(format!( + "waited {elapsed_str}; no incoming event arrived; current time: {current_time}" + )) + } + interrupt = rx.recv() => { + let elapsed = start_time.elapsed(); + let elapsed_str = format_duration_human(elapsed); + let current_time = chrono::Local::now().format("%Y-%m-%d %H:%M:%S %Z").to_string(); + match interrupt { + Some(Interrupt::Reset) => RuntimeWaitResult::Reset, + Some(Interrupt::Compact) => RuntimeWaitResult::Compact, + Some(Interrupt::UpdateSoul { content }) => RuntimeWaitResult::UpdateSoul(content), + Some(Interrupt::DebugReflectRequest) => RuntimeWaitResult::Timeout( + format!("debug reflection request received while waiting; current time: {current_time}; elapsed since last turn: {elapsed_str}; try again after this turn") + ), + Some(interrupt) => { + let source = interrupt.source_tag().to_string(); + RuntimeWaitResult::Incoming { + result: format!( + "interrupted early by incoming {source} event after {elapsed_str}; current time: {current_time}" + ), + interrupt, + } + } + None => RuntimeWaitResult::Timeout(format!("interrupt channel closed; current time: {current_time}; elapsed since last turn: {elapsed_str}; continue")), + } + } + } +} + +pub fn wait_timeout_without_event(result: &str) -> bool { + result.contains("no incoming event arrived") +} + +pub fn format_duration_human(d: Duration) -> String { + let secs = d.as_secs(); + if secs == 0 { + return format!("{}ms", d.as_millis()); + } + let h = secs / 3600; + let m = (secs % 3600) / 60; + let s = secs % 60; + let mut parts = vec![]; + if h > 0 { + parts.push(format!("{h}hour")); + } + if m > 0 { + parts.push(format!("{m}min")); + } + if s > 0 || parts.is_empty() { + parts.push(format!("{s}sec")); + } + parts.join(" ") +} + #[cfg(test)] mod tests { use std::time::Duration; - use super::timeout_duration; + use super::{format_duration_human, timeout_duration}; #[test] fn timeout_duration_clamps_to_bounds() { @@ -98,5 +183,16 @@ Duration::from_millis(600_000) ); assert!(timeout_duration(&serde_json::json!({})).is_err()); + } + + #[test] + fn format_duration_human_keeps_compact_units() { + assert_eq!(format_duration_human(Duration::from_millis(500)), "500ms"); + assert_eq!(format_duration_human(Duration::from_secs(5)), "5sec"); + assert_eq!(format_duration_human(Duration::from_secs(65)), "1min 5sec"); + assert_eq!( + format_duration_human(Duration::from_secs(3665)), + "1hour 1min 5sec" + ); } } -- tangled.sh