Compare commits

...
20 Commits
Author SHA1 Message Date
Timmy 69cf867969 Bump version to 0.14.5
Publish sled artifact / publish-dev-artifact (push) Canceled after 0s
2026-07-24 17:41:05 +01:00
Huskies Agent b43def12f7 huskies: regen source-map.json 2026-07-21 20:19:29 +00:00
Huskies Agent c3cad6b3d8 huskies: merge 1244 refactor Deduplicate the worktree module cluster (cleanup/remove/sweep/create/lifecycle share ~60-line clones) 2026-07-21 20:19:29 +00:00
Huskies Agent c58e204e8b huskies: regen source-map.json 2026-07-21 17:51:35 +00:00
Huskies Agent 8b2dd21e22 huskies: merge 1247 refactor Deduplicate chat-transport LLM command handling (discord <-> whatsapp 45-line clone) 2026-07-21 17:51:35 +00:00
Huskies Agent 326b2b4a32 huskies: regen source-map.json 2026-07-21 17:29:35 +00:00
Huskies Agent 10dd239c92 huskies: merge 1246 refactor Extract shared LLM-runtime logic duplicated between gemini and openai runtimes 2026-07-21 17:29:35 +00:00
Huskies Agent 1583ade9fc huskies: merge 1245 refactor Deduplicate validation/requests.rs — five internal ~50-85 line self-clones 2026-07-21 16:55:10 +00:00
Huskies Agent 5c3b433e7d huskies: merge 1251 story script/check gates duplication and cognitive complexity 2026-07-21 16:15:10 +00:00
Huskies Agent 14331127b8 huskies: regen source-map.json 2026-07-21 16:08:03 +00:00
Huskies Agent 7de8fa1924 huskies: merge 1249 story Checkpoint the CRDT snapshot periodically, not once by accident 2026-07-21 16:08:03 +00:00
Huskies Agent 82d9e3c460 huskies: regen source-map.json 2026-07-21 15:25:35 +00:00
Huskies Agent 013a5da60b huskies: merge 1243 refactor Merge jobs stop bloating replicated CRDT state 2026-07-21 15:25:34 +00:00
Huskies Agent 2ea633a2f1 huskies: merge 1242 refactor script/release builds project images so they can't drift 2026-07-21 15:10:39 +00:00
Huskies Agent b9e0a16bf8 huskies: regen source-map.json 2026-07-21 14:55:49 +00:00
Huskies Agent 78b2e7a9a2 huskies: merge 1250 bug Bug fields steps_to_reproduce, actual_result and expected_result cannot be edited after creation 2026-07-21 14:55:49 +00:00
Huskies Agent d06f5b5410 huskies: merge 1241 bug CRDT snapshot has no schema migration; a failed load silently starts empty 2026-07-21 14:47:40 +00:00
Timmy 40fb6367c1 Bump version to 0.14.4
Publish sled artifact / publish-dev-artifact (push) Canceled after 0s
2026-07-21 14:51:39 +01:00
Huskies Agent 1859e79491 huskies: regen source-map.json 2026-07-21 13:46:43 +00:00
Huskies Agent 5eeb036875 huskies: merge 1248 bug Duplicate Working line: obsolete digging-in watcher survives alongside 1240 placeholder 2026-07-21 13:46:43 +00:00
57 changed files with 1925 additions and 2559 deletions
+44 -4
View File
@@ -767,6 +767,25 @@
"server/src/agents/pty/types.rs": [
"struct PtyResult"
],
"server/src/agents/runtime/api_common.rs": [
"struct CancellationFlag",
"fn new",
"fn handle",
"fn stop",
"fn status",
"fn make_emit",
"fn zero_usage",
"fn emit_running",
"fn start_conversation_loop",
"fn api_runtime_result",
"fn done_result",
"fn check_loop_guard",
"fn extract_model",
"fn build_system_text",
"fn clean_schema_properties",
"fn test_app_ctx",
"fn test_runtime_context"
],
"server/src/agents/runtime/claude_code.rs": [
"struct ClaudeCodeRuntime",
"fn new"
@@ -950,6 +969,12 @@
"fn load_chat_history",
"fn save_chat_history"
],
"server/src/chat/llm_turn.rs": [
"trait TransportFormat",
"fn build_llm_prompt",
"struct LlmTurnContext",
"fn run_llm_turn"
],
"server/src/chat/lookup.rs": [
"fn find_story_by_number"
],
@@ -958,6 +983,7 @@
"mod compact",
"mod dispatcher",
"mod history",
"mod llm_turn",
"mod lookup",
"mod test_helpers",
"mod timer",
@@ -1042,7 +1068,6 @@
],
"server/src/chat/transport/matrix/bot/messages/handle_message.rs": [
"const DIGGING_IN_MESSAGE",
"fn spawn_digging_in_watcher",
"fn handle_message"
],
"server/src/chat/transport/matrix/bot/messages/mod.rs": [
@@ -1103,7 +1128,6 @@
"fn default_compact_seed_max_bytes",
"fn default_cache_read_suggest_threshold",
"fn default_compact_suggest_cooldown_secs",
"fn default_digging_in_threshold_secs",
"fn default_transport",
"fn default_whatsapp_provider",
"struct BotConfig"
@@ -1354,6 +1378,7 @@
"mod agent_name",
"struct ProjectConfig",
"struct WatcherConfig",
"fn default_snapshot_interval_ops",
"struct DiskWatchConfig",
"struct ComponentConfig",
"struct AgentConfig",
@@ -1514,7 +1539,8 @@
"server/src/crdt_state/state/init.rs": [
"enum PersistMsg",
"fn init",
"fn flush_persistence"
"fn flush_persistence",
"fn checkpoint_on_shutdown"
],
"server/src/crdt_state/state/mod.rs": [
"fn subscribe",
@@ -1530,6 +1556,10 @@
"static SYNC_TX",
"static ALL_OPS",
"static VECTOR_CLOCK",
"static CRDT_POOL",
"static SNAPSHOT_INTERVAL_OPS",
"static OPS_SINCE_SNAPSHOT",
"static LAST_SEQ",
"static ALL_OPS_TL",
"static VECTOR_CLOCK_TL",
"fn all_ops_lock",
@@ -2074,6 +2104,7 @@
],
"server/src/http/workflow/bug_ops/bug.rs": [
"fn create_bug_file",
"fn update_bug_fields_in_file",
"fn is_bug_item",
"fn extract_bug_name_from_content",
"fn list_bug_files"
@@ -2892,7 +2923,10 @@
"const SUMMARY_TAIL_LINES",
"fn extract_failing_test_names",
"fn summarize_merge_result",
"fn summarize_merge_failure_kind"
"fn summarize_merge_failure_kind",
"fn bound_result_for_storage",
"fn bound_report_for_storage",
"fn bound_plain_error"
],
"server/src/service/mod.rs": [
"mod agents",
@@ -3515,6 +3549,7 @@
"fn migrate_slug_paths"
],
"server/src/worktree/mod.rs": [
"mod test_support",
"struct WorktreeInfo",
"struct WorktreeListEntry",
"fn worktree_path",
@@ -3529,5 +3564,10 @@
"server/src/worktree/sweep.rs": [
"fn worktree_should_be_swept",
"fn sweep_with_lookup"
],
"server/src/worktree/test_support.rs": [
"fn init_git_repo",
"fn empty_config",
"fn done_stage"
]
}
+12
View File
@@ -0,0 +1,12 @@
{
"threshold": 10,
"minLines": 10,
"minTokens": 50,
"ignore": [
"**/target/**",
"**/node_modules/**",
"**/dist/**",
"**/*.svg",
"**/flamegraphs/**"
]
}
Generated
+19 -19
View File
@@ -67,9 +67,9 @@ checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
[[package]]
name = "ammonia"
version = "4.1.3"
version = "4.1.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "68b9d3370580a12f4b7a10fdcc18b28942c083ba570e3d954fe59d10951b85a2"
checksum = "dc6d763210e2eb7670d1a5183a08bebefa3f97db2a738a684f2ce00bd49f681d"
dependencies = [
"cssparser",
"html5ever",
@@ -227,7 +227,7 @@ checksum = "ae36dc4177970ef04fde5178d3e2429882def40e57a451f919c098f72baa6cec"
dependencies = [
"proc-macro2",
"quote",
"syn 3.0.2",
"syn 3.0.3",
]
[[package]]
@@ -456,9 +456,9 @@ dependencies = [
[[package]]
name = "cc"
version = "1.3.0"
version = "1.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c89588d05638b5b4594a3348a2d6c20277e43a7f5c5202b05cc56888475a47b8"
checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -583,9 +583,9 @@ dependencies = [
[[package]]
name = "clap"
version = "4.6.3"
version = "4.6.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0fb99565819980999fb7b4a1796046a5c949e6d4ff132cf5fadf5a641e20d776"
checksum = "d91e0c145792ef73a6ad36d27c75ac09f1832222a3c209689d90f534685ee5b7"
dependencies = [
"clap_builder",
]
@@ -1925,7 +1925,7 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
[[package]]
name = "huskies"
version = "0.14.3"
version = "0.14.5"
dependencies = [
"ammonia",
"async-stream",
@@ -2492,9 +2492,9 @@ checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
[[package]]
name = "libc"
version = "0.2.188"
version = "0.2.189"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22053b6a34f84abc97f9129e61334f40174659a1b9bd18c970b83db6a9a6348b"
checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
[[package]]
name = "libsqlite3-sys"
@@ -3738,7 +3738,7 @@ checksum = "2c9283685feec7d69af75fb0e858d5e7378f33fe4fc699383b2916ab9273e03c"
dependencies = [
"proc-macro2",
"quote",
"syn 3.0.2",
"syn 3.0.3",
]
[[package]]
@@ -4080,9 +4080,9 @@ dependencies = [
[[package]]
name = "rustls-pki-types"
version = "1.15.0"
version = "1.15.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "764899a24af3980067ee14bc143654f297b22eaebfe3c7b6b211920a5a59b046"
checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96"
dependencies = [
"web-time",
"zeroize",
@@ -4275,7 +4275,7 @@ checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
dependencies = [
"proc-macro2",
"quote",
"syn 3.0.2",
"syn 3.0.3",
]
[[package]]
@@ -4845,9 +4845,9 @@ dependencies = [
[[package]]
name = "syn"
version = "3.0.2"
version = "3.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a207d6d6a2b7fc470b80443726053f18a2481b7e1eee970597051596567987a3"
checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3"
dependencies = [
"proc-macro2",
"quote",
@@ -4954,7 +4954,7 @@ checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd"
dependencies = [
"proc-macro2",
"quote",
"syn 3.0.2",
"syn 3.0.3",
]
[[package]]
@@ -5071,9 +5071,9 @@ dependencies = [
[[package]]
name = "tokio-stream"
version = "0.1.18"
version = "0.1.19"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "32da49809aab5c3bc678af03902d4ccddea2a87d028d86392a4b1560c6906c70"
checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b"
dependencies = [
"futures-core",
"pin-project-lite",
+7
View File
@@ -0,0 +1,7 @@
# cognitive_complexity is allow-by-default in clippy; script/check enables it
# with `-W clippy::cognitive_complexity`. This threshold is set well above
# clippy's own default (25) to accommodate existing large dispatch functions
# (e.g. Matrix bot command routing) without requiring an unrelated refactor;
# it still gates against genuinely runaway complexity introduced going
# forward.
cognitive-complexity-threshold = 200
+5
View File
@@ -88,6 +88,11 @@ RUN curl -fsSL https://deb.nodesource.com/setup_22.x | bash - \
# Claude Code CLI in runtime
RUN npm install -g @anthropic-ai/claude-code
# jscpd — duplication detector used by script/check. Installed in the
# runtime stage (not just the base build stage) so it's available to agents
# running script/check inside the sled, not only on a developer machine.
RUN npm install -g jscpd
# Cargo and Rust toolchain needed at runtime for:
# - rebuild_and_restart (cargo build inside the container)
# - Agent-driven cargo commands (cargo clippy, cargo test, etc.)
+2 -2
View File
@@ -1,12 +1,12 @@
{
"name": "huskies",
"version": "0.14.3",
"version": "0.14.5",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "huskies",
"version": "0.14.3",
"version": "0.14.5",
"dependencies": {
"@types/react-syntax-highlighter": "^15.5.13",
"react": "^19.1.0",
+1 -1
View File
@@ -1,7 +1,7 @@
{
"name": "huskies",
"private": true,
"version": "0.14.3",
"version": "0.14.5",
"type": "module",
"scripts": {
"dev": "vite",
+4 -13
View File
@@ -4,19 +4,10 @@ set -euo pipefail
# Build all project images in dependency order:
# huskies → huskies-project-base → huskies-project-<stack> (one per stack fragment)
#
# Run this after `script/docker_rebuild` or whenever you add a new stack.
# Safe to re-run: each step re-tags the image with the latest layers.
#
# IMPORTANT (story 1231): this script is NOT part of `script/release`. The
# huskies-project-* images bake whatever `huskies` binary happened to be built
# locally the last time this ran — as of this comment that's 0.13.0, several
# releases behind the current fleet artifact. `project-rebuild` self-heals a
# sled that comes back on a stale baked binary (it re-upgrades it in place
# from the gateway's published artifact), but that costs an extra
# download+restart cycle every time. Run `script/build-project-images` after
# every `script/release` — ideally as a step in the release flow itself — so
# freshly rebuilt sleds start on a current binary instead of relying on the
# self-heal.
# Called automatically by `script/release` (story 1242) so the huskies-project-*
# images never drift from the version being published. Also safe to run
# standalone after `script/docker_rebuild` or whenever you add a new stack —
# each step re-tags the image with the latest layers.
cd "$(dirname "$0")/.."
+18 -4
View File
@@ -1,7 +1,8 @@
#!/usr/bin/env bash
# Pre-commit quality gate: fmt-check, clippy, cargo check, and doc-coverage.
# Run this before committing to catch fmt drift, clippy warnings, compile
# errors, and missing doc comments without waiting for the full test suite.
# Pre-commit quality gate: fmt-check, clippy, duplication, cargo check, and
# doc-coverage. Run this before committing to catch fmt drift, clippy
# warnings, duplicate code, compile errors, and missing doc comments without
# waiting for the full test suite.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
@@ -11,7 +12,20 @@ echo "=== Checking Rust formatting ==="
cargo fmt --manifest-path "$PROJECT_ROOT/Cargo.toml" --all --check
echo "=== Running cargo clippy ==="
cargo clippy --manifest-path "$PROJECT_ROOT/Cargo.toml" --workspace --all-targets -- -D warnings
# clippy::cognitive_complexity is allow-by-default; -W activates it so that
# -D warnings turns violations into a hard failure. The threshold it's
# measured against lives in clippy.toml (committed, not passed ad hoc).
cargo clippy --manifest-path "$PROJECT_ROOT/Cargo.toml" --workspace --all-targets -- -W clippy::cognitive_complexity -D warnings
echo "=== Checking code duplication (jscpd) ==="
if ! command -v jscpd &>/dev/null; then
echo "FAIL: jscpd is not installed. Install it with 'npm install -g jscpd' before running script/check." >&2
exit 1
fi
# Duplication threshold lives in .jscpd.json (committed, not passed ad hoc).
# jscpd exits non-zero automatically when duplication exceeds that threshold.
jscpd --config "$PROJECT_ROOT/.jscpd.json" \
"$PROJECT_ROOT/server/src" "$PROJECT_ROOT/frontend/src" "$PROJECT_ROOT/crates"
echo "=== Checking doc coverage on changed files ==="
cargo run --manifest-path "$PROJECT_ROOT/Cargo.toml" -p source-map-gen --bin source-map-check --quiet -- --worktree "$PROJECT_ROOT" --base master
+13 -10
View File
@@ -1,16 +1,6 @@
#!/usr/bin/env bash
set -euo pipefail
# NOTE (story 1231): this script publishes the `huskies` binary artifact
# (consumed by the `upgrade`/`upgrade all` chat commands) but does NOT
# rebuild the huskies-project-* Docker images that `project-rebuild` uses.
# Those images bake whatever binary was locally built the last time
# `script/build-project-images` ran, so they silently drift stale otherwise
# (currently baking 0.13.0). Run `script/build-project-images` after this
# script — or better, fold it into this release flow — so a `project-rebuild`
# right after a release doesn't need to self-heal a stale sled back up to the
# version just published here.
# ── Configuration ──────────────────────────────────────────────
GITEA_URL="https://code.crashlabs.io"
REPO="crashlabs/huskies"
@@ -97,6 +87,19 @@ cross build --release --target x86_64-unknown-linux-musl
echo "==> Building Linux arm64 (static musl via cross)..."
cross build --release --target aarch64-unknown-linux-musl
# ── Build project images ─────────────────────────────────────────
# Rebuild the huskies-project-* Docker images from this exact source tree
# (the version-bump commit above already landed, so build.rs's `git
# rev-parse HEAD` embeds the matching git hash) so they never drift from
# the binary being published below. A release that can't produce these
# images fails loudly here, before anything is tagged, pushed, or published.
echo "==> Building project images..."
if ! "${SCRIPT_DIR}/script/build-project-images"; then
echo "Error: failed to build huskies-project-* images at ${VERSION}."
echo "Release aborted — nothing was tagged, pushed, or published."
exit 1
fi
# ── Package ────────────────────────────────────────────────────
DIST="target/dist"
rm -rf "$DIST"
+1 -1
View File
@@ -1,6 +1,6 @@
[package]
name = "huskies"
version = "0.14.3"
version = "0.14.5"
edition = "2024"
build = "build.rs"
@@ -299,10 +299,17 @@ impl AgentPool {
crate::db::write_content(crate::db::ContentKey::MergeSuccess(&sid), "1");
}
// Update CRDT with terminal status.
// Update CRDT with terminal status. The full untruncated output is
// already on disk (write_merge_report, called from
// run_merge_pipeline for the Ok(r) case above, or below for the
// Err(e) case) — only a bounded summary plus that pointer goes
// into the replicated `merge_jobs.error` field, so a large gate
// failure doesn't bloat every node's CRDT state.
match &report {
Ok(r) => {
let report_json = serde_json::to_string(r).unwrap_or_else(|_| String::new());
let bounded = crate::service::merge::bound_report_for_storage(r);
let report_json =
serde_json::to_string(&bounded).unwrap_or_else(|_| String::new());
crate::crdt_state::write_merge_job(
&sid,
"completed",
@@ -312,12 +319,15 @@ impl AgentPool {
);
}
Err(e) => {
let report_path = crate::service::merge::io::write_merge_report(&root, &sid, e);
let bounded =
crate::service::merge::bound_plain_error(e, report_path.as_deref());
crate::crdt_state::write_merge_job(
&sid,
"failed",
started_at,
Some(finished_at),
Some(e),
Some(&bounded),
);
}
}
+1 -29
View File
@@ -184,39 +184,11 @@ pub(crate) async fn on_terminal_transition(project_root: &Path, story_id: &str)
#[cfg(test)]
mod tests {
use super::*;
use crate::worktree::test_support::init_git_repo;
use std::fs;
use std::process::Command;
use tempfile::TempDir;
fn init_git_repo(dir: &Path) {
use crate::git_test_support::git_ok;
git_ok(
Command::new("git").args(["init"]).current_dir(dir).output(),
"git init",
);
git_ok(
Command::new("git")
.args(["config", "user.email", "test@test.com"])
.current_dir(dir)
.output(),
"git config user.email",
);
git_ok(
Command::new("git")
.args(["config", "user.name", "Test"])
.current_dir(dir)
.output(),
"git config user.name",
);
git_ok(
Command::new("git")
.args(["commit", "--allow-empty", "-m", "init"])
.current_dir(dir)
.output(),
"git commit",
);
}
fn setup_project(tmp: &TempDir) -> PathBuf {
let root = tmp.path().join("project");
fs::create_dir_all(root.join(".huskies")).unwrap();
+404
View File
@@ -0,0 +1,404 @@
//! Shared helpers for the API-based agent runtimes (Gemini, OpenAI), which
//! talk directly to a provider's REST API rather than spawning a CLI over a
//! PTY. Both runtimes drive an almost-identical turn loop against different
//! wire formats; this module holds the logic that doesn't vary between them.
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use serde_json::Value;
use tokio::sync::broadcast;
use crate::agent_log::AgentLogWriter;
use super::super::{AgentEvent, TokenUsage};
use super::{RuntimeContext, RuntimeResult, RuntimeStatus};
/// Cancellation flag shared by the API-based runtimes' `stop()`/`get_status()`.
pub(super) struct CancellationFlag {
cancelled: Arc<AtomicBool>,
}
impl CancellationFlag {
/// Create a fresh, un-cancelled flag.
pub(super) fn new() -> Self {
Self {
cancelled: Arc::new(AtomicBool::new(false)),
}
}
/// Clone of the underlying flag, for the conversation loop to poll.
pub(super) fn handle(&self) -> Arc<AtomicBool> {
Arc::clone(&self.cancelled)
}
/// Request a stop.
pub(super) fn stop(&self) {
self.cancelled.store(true, Ordering::Relaxed);
}
/// Report `Failed` once stopped, `Idle` otherwise.
pub(super) fn status(&self) -> RuntimeStatus {
if self.cancelled.load(Ordering::Relaxed) {
RuntimeStatus::Failed
} else {
RuntimeStatus::Idle
}
}
}
/// Build the event-emitting closure shared by API-based runtimes: forwards
/// events to the broadcast channel, the in-memory event log, and (optionally)
/// the on-disk log writer.
pub(super) fn make_emit(
tx: broadcast::Sender<AgentEvent>,
event_log: Arc<Mutex<Vec<AgentEvent>>>,
log_writer: Option<Arc<Mutex<AgentLogWriter>>>,
) -> impl Fn(AgentEvent) {
move |event: AgentEvent| {
super::super::pty::emit_event(
event,
&tx,
&event_log,
log_writer.as_ref().map(|w| w.as_ref()),
);
}
}
/// Zeroed token-usage accumulator, seeded before an API runtime's
/// conversation loop starts accumulating per-turn usage.
pub(super) fn zero_usage() -> TokenUsage {
TokenUsage {
input_tokens: 0,
output_tokens: 0,
cache_creation_input_tokens: 0,
cache_read_input_tokens: 0,
total_cost_usd: 0.0,
}
}
/// Emit the initial "running" status event.
pub(super) fn emit_running(ctx: &RuntimeContext, emit: &impl Fn(AgentEvent)) {
emit(AgentEvent::Status {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
status: "running".to_string(),
});
}
/// Set up an API-based runtime's conversation loop: builds the event
/// emitter, emits the initial "running" status, and returns it alongside a
/// zeroed usage accumulator and the turn counter (starting at 0).
pub(super) fn start_conversation_loop(
ctx: &RuntimeContext,
tx: broadcast::Sender<AgentEvent>,
event_log: Arc<Mutex<Vec<AgentEvent>>>,
log_writer: Option<Arc<Mutex<AgentLogWriter>>>,
) -> (impl Fn(AgentEvent), TokenUsage, u32) {
let emit = make_emit(tx, event_log, log_writer);
emit_running(ctx, &emit);
(emit, zero_usage(), 0u32)
}
/// Build a successful `RuntimeResult` carrying the given token usage. All
/// API-based runtimes report `exit_ok: true` and leave the CLI-only fields
/// (`aborted_signal`, `rate_limit_exit`, `rate_limit_reset_at`) at their
/// defaults, since those concepts don't apply outside the PTY runtime.
pub(super) fn api_runtime_result(total_usage: TokenUsage) -> RuntimeResult {
RuntimeResult {
session_id: None,
token_usage: Some(total_usage),
exit_ok: true,
aborted_signal: false,
rate_limit_exit: false,
rate_limit_reset_at: None,
}
}
/// Emit the `Done` event and build the final successful result once the
/// model produces a response with no further tool/function calls.
pub(super) fn done_result(
ctx: &RuntimeContext,
emit: &impl Fn(AgentEvent),
total_usage: TokenUsage,
) -> RuntimeResult {
emit(AgentEvent::Done {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
session_id: None,
});
api_runtime_result(total_usage)
}
/// Safety limit on conversation turns for API-based runtimes, shared so
/// both the guard check and its error message stay in sync.
const MAX_TURNS: u32 = 200;
/// Check the per-turn cancellation/max-turns guard at the top of an API
/// runtime's conversation loop. Returns `Some(result)` when the loop should
/// stop immediately (either the user requested a stop, or the safety turn
/// limit was exceeded); otherwise increments `*turn` and returns `None`.
pub(super) fn check_loop_guard(
ctx: &RuntimeContext,
cancelled: &AtomicBool,
turn: &mut u32,
total_usage: &TokenUsage,
emit: &impl Fn(AgentEvent),
) -> Option<RuntimeResult> {
if cancelled.load(Ordering::Relaxed) {
emit(AgentEvent::Error {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
message: "Agent was stopped by user".to_string(),
});
return Some(api_runtime_result(total_usage.clone()));
}
*turn += 1;
if *turn > MAX_TURNS {
emit(AgentEvent::Error {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
message: format!("Exceeded maximum turns ({MAX_TURNS})"),
});
return Some(api_runtime_result(total_usage.clone()));
}
None
}
/// Extract the model name for an API-based runtime: the agent pool stashes
/// the model directly in `ctx.command` for non-CLI runtimes, detected here
/// via `is_command_a_model`; otherwise fall back to a `--model` arg, and
/// finally `default_model`.
pub(super) fn extract_model(
ctx: &RuntimeContext,
is_command_a_model: impl Fn(&str) -> bool,
default_model: &str,
) -> String {
if is_command_a_model(&ctx.command) {
ctx.command.clone()
} else {
ctx.args
.iter()
.position(|a| a == "--model")
.and_then(|i| ctx.args.get(i + 1))
.cloned()
.unwrap_or_else(|| default_model.to_string())
}
}
/// Build the default system-prompt text shared by both API-based runtimes:
/// prefers an explicit `--append-system-prompt` arg (set by the agent pool),
/// else falls back to a generic tool-calling preamble.
pub(super) fn build_system_text(ctx: &RuntimeContext) -> String {
ctx.args
.iter()
.position(|a| a == "--append-system-prompt")
.and_then(|i| ctx.args.get(i + 1))
.cloned()
.unwrap_or_else(|| {
format!(
"You are an AI coding agent working on story {}. \
You have access to tools via function calling. \
Use them to complete the task. \
Work in the directory: {}",
ctx.story_id, ctx.cwd
)
})
}
/// Recursively clean an MCP JSON-Schema `properties` object into a provider
/// function-calling schema: strips `$schema` (always) and
/// `additionalProperties` (when the provider doesn't support it, e.g.
/// Gemini) from the top level and from nested `properties`/`items`.
pub(super) fn clean_schema_properties(
properties: &Value,
strip_additional_properties: bool,
) -> Value {
let Some(obj) = properties.as_object() else {
return properties.clone();
};
let mut cleaned = serde_json::Map::new();
for (key, value) in obj {
let mut prop = value.clone();
if let Some(p) = prop.as_object_mut() {
p.remove("$schema");
if strip_additional_properties {
p.remove("additionalProperties");
}
if let Some(nested_props) = p.get("properties").cloned() {
p.insert(
"properties".to_string(),
clean_schema_properties(&nested_props, strip_additional_properties),
);
}
if let Some(items) = p.get("items").cloned()
&& let Some(items_obj) = items.as_object()
{
let mut cleaned_items = items_obj.clone();
cleaned_items.remove("$schema");
if strip_additional_properties {
cleaned_items.remove("additionalProperties");
}
p.insert("items".to_string(), Value::Object(cleaned_items));
}
}
cleaned.insert(key.clone(), prop);
}
Value::Object(cleaned)
}
// ── Test helpers ─────────────────────────────────────────────────────
/// Build a throwaway `AppContext` backed by a temp directory, for tests
/// that need a `RuntimeContext.app_ctx` but don't exercise it.
#[cfg(test)]
pub(super) fn test_app_ctx() -> Arc<crate::http::context::AppContext> {
let tmp = tempfile::tempdir().unwrap();
Arc::new(crate::http::context::AppContext::new_test(
tmp.path().to_path_buf(),
))
}
/// Build a `RuntimeContext` with sensible test defaults, overriding only
/// `command` and `args` (the fields the API-runtime tests vary).
#[cfg(test)]
pub(super) fn test_runtime_context(command: &str, args: Vec<String>) -> RuntimeContext {
RuntimeContext {
story_id: "42_story_test".to_string(),
agent_name: "coder-1".to_string(),
command: command.to_string(),
args,
prompt: "Do the thing".to_string(),
cwd: "/tmp/wt".to_string(),
inactivity_timeout_secs: 300,
app_ctx: Some(test_app_ctx()),
session_id_to_resume: None,
fresh_prompt: None,
project_root: std::path::PathBuf::from("/tmp/project"),
model: None,
}
}
// ── Tests ────────────────────────────────────────────────────────────
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn clean_schema_strips_dollar_schema_always() {
let schema = json!({
"name": { "type": "string", "$schema": "http://json-schema.org/draft-07/schema#" }
});
let result = clean_schema_properties(&schema, false);
assert!(result["name"].get("$schema").is_none());
}
#[test]
fn clean_schema_strips_additional_properties_when_requested() {
let schema = json!({
"name": { "type": "string", "additionalProperties": false }
});
let result = clean_schema_properties(&schema, true);
assert!(result["name"].get("additionalProperties").is_none());
}
#[test]
fn clean_schema_keeps_additional_properties_when_not_requested() {
let schema = json!({
"name": { "type": "object", "additionalProperties": false }
});
let result = clean_schema_properties(&schema, false);
assert!(result["name"].get("additionalProperties").is_some());
}
#[test]
fn clean_schema_recurses_into_nested_object_properties() {
let schema = json!({
"config": {
"type": "object",
"properties": {
"key": { "type": "string", "$schema": "x" }
}
}
});
let result = clean_schema_properties(&schema, false);
assert!(result["config"]["properties"]["key"].is_object());
assert!(
result["config"]["properties"]["key"]
.get("$schema")
.is_none()
);
}
#[test]
fn clean_schema_recurses_into_array_items() {
let schema = json!({
"items": {
"type": "array",
"items": {
"type": "object",
"properties": { "name": { "type": "string" } },
"additionalProperties": false,
"$schema": "x"
}
}
});
let result = clean_schema_properties(&schema, true);
let items_schema = &result["items"]["items"];
assert!(items_schema.get("additionalProperties").is_none());
assert!(items_schema.get("$schema").is_none());
}
#[test]
fn extract_model_uses_command_when_it_matches() {
let ctx = test_runtime_context("gpt-4o", vec![]);
assert_eq!(
extract_model(&ctx, |c| c.starts_with("gpt"), "fallback"),
"gpt-4o"
);
}
#[test]
fn extract_model_falls_back_to_args() {
let ctx = test_runtime_context("claude", vec!["--model".to_string(), "custom".to_string()]);
assert_eq!(
extract_model(&ctx, |c| c.starts_with("gpt"), "fallback"),
"custom"
);
}
#[test]
fn extract_model_falls_back_to_default() {
let ctx = test_runtime_context("claude", vec![]);
assert_eq!(
extract_model(&ctx, |c| c.starts_with("gpt"), "fallback"),
"fallback"
);
}
#[test]
fn build_system_text_uses_args() {
let ctx = test_runtime_context(
"gpt-4o",
vec![
"--append-system-prompt".to_string(),
"Custom system prompt".to_string(),
],
);
assert_eq!(build_system_text(&ctx), "Custom system prompt");
}
#[test]
fn build_system_text_default() {
let ctx = test_runtime_context("gpt-4o", vec![]);
let text = build_system_text(&ctx);
assert!(text.contains("42_story_test"));
assert!(text.contains("/tmp/wt"));
}
}
+8 -54
View File
@@ -4,6 +4,7 @@ use serde_json::{Value, json};
use super::super::super::TokenUsage;
use super::super::RuntimeContext;
use super::super::api_common::build_system_text;
// ── Gemini API types ─────────────────────────────────────────────────
@@ -19,26 +20,8 @@ pub(super) struct GeminiFunctionDeclaration {
/// Build the system instruction content from the RuntimeContext.
pub(super) fn build_system_instruction(ctx: &RuntimeContext) -> Value {
// Use system_prompt from args if provided via --append-system-prompt,
// otherwise use a sensible default.
let system_text = ctx
.args
.iter()
.position(|a| a == "--append-system-prompt")
.and_then(|i| ctx.args.get(i + 1))
.cloned()
.unwrap_or_else(|| {
format!(
"You are an AI coding agent working on story {}. \
You have access to tools via function calling. \
Use them to complete the task. \
Work in the directory: {}",
ctx.story_id, ctx.cwd
)
});
json!({
"parts": [{ "text": system_text }]
"parts": [{ "text": build_system_text(ctx) }]
})
}
@@ -92,34 +75,18 @@ pub(super) fn parse_usage_metadata(response: &Value) -> Option<TokenUsage> {
#[cfg(test)]
mod tests {
use super::super::super::api_common::test_runtime_context;
use super::*;
use crate::http::context::AppContext;
use std::sync::Arc;
fn test_app_ctx() -> Arc<AppContext> {
let tmp = tempfile::tempdir().unwrap();
Arc::new(AppContext::new_test(tmp.path().to_path_buf()))
}
#[test]
fn build_system_instruction_uses_args() {
let ctx = RuntimeContext {
story_id: "42_story_test".to_string(),
agent_name: "coder-1".to_string(),
command: "gemini-2.5-pro".to_string(),
args: vec![
let ctx = test_runtime_context(
"gemini-2.5-pro",
vec![
"--append-system-prompt".to_string(),
"Custom system prompt".to_string(),
],
prompt: "Do the thing".to_string(),
cwd: "/tmp/wt".to_string(),
inactivity_timeout_secs: 300,
app_ctx: Some(test_app_ctx()),
session_id_to_resume: None,
fresh_prompt: None,
project_root: std::path::PathBuf::from("/tmp/project"),
model: None,
};
);
let instruction = build_system_instruction(&ctx);
assert_eq!(instruction["parts"][0]["text"], "Custom system prompt");
@@ -127,20 +94,7 @@ mod tests {
#[test]
fn build_system_instruction_default() {
let ctx = RuntimeContext {
story_id: "42_story_test".to_string(),
agent_name: "coder-1".to_string(),
command: "gemini-2.5-pro".to_string(),
args: vec![],
prompt: "Do the thing".to_string(),
cwd: "/tmp/wt".to_string(),
inactivity_timeout_secs: 300,
app_ctx: Some(test_app_ctx()),
session_id_to_resume: None,
fresh_prompt: None,
project_root: std::path::PathBuf::from("/tmp/project"),
model: None,
};
let ctx = test_runtime_context("gemini-2.5-pro", vec![]);
let instruction = build_system_instruction(&ctx);
let text = instruction["parts"][0]["text"].as_str().unwrap();
+2 -80
View File
@@ -8,6 +8,7 @@ use crate::slog;
use crate::http::mcp::tools_list::list_tools;
use super::super::api_common::clean_schema_properties;
use super::api::GeminiFunctionDeclaration;
// ── MCP tool loading ────────────────────────────────────────────────
@@ -62,7 +63,7 @@ pub(super) fn convert_mcp_schema_to_gemini(schema: Option<&Value>) -> Option<Val
let mut result = json!({
"type": "object",
"properties": clean_schema_properties(properties),
"properties": clean_schema_properties(properties, true),
});
// Preserve required fields if present.
@@ -73,44 +74,6 @@ pub(super) fn convert_mcp_schema_to_gemini(schema: Option<&Value>) -> Option<Val
Some(result)
}
/// Recursively clean schema properties to be Gemini-compatible.
/// Removes unsupported JSON Schema keywords.
fn clean_schema_properties(properties: &Value) -> Value {
let Some(obj) = properties.as_object() else {
return properties.clone();
};
let mut cleaned = serde_json::Map::new();
for (key, value) in obj {
let mut prop = value.clone();
// Remove JSON Schema keywords not supported by Gemini
if let Some(p) = prop.as_object_mut() {
p.remove("$schema");
p.remove("additionalProperties");
// Recursively clean nested object properties
if let Some(nested_props) = p.get("properties").cloned() {
p.insert(
"properties".to_string(),
clean_schema_properties(&nested_props),
);
}
// Clean items schema for arrays
if let Some(items) = p.get("items").cloned()
&& let Some(items_obj) = items.as_object()
{
let mut cleaned_items = items_obj.clone();
cleaned_items.remove("$schema");
cleaned_items.remove("additionalProperties");
p.insert("items".to_string(), Value::Object(cleaned_items));
}
}
cleaned.insert(key.clone(), prop);
}
Value::Object(cleaned)
}
// ── Tests ────────────────────────────────────────────────────────────
#[cfg(test)]
@@ -170,45 +133,4 @@ mod tests {
assert!(name_prop.get("$schema").is_none());
assert_eq!(name_prop["type"], "string");
}
#[test]
fn convert_mcp_schema_with_nested_objects() {
let schema = json!({
"type": "object",
"properties": {
"config": {
"type": "object",
"properties": {
"key": { "type": "string" }
}
}
}
});
let result = convert_mcp_schema_to_gemini(Some(&schema)).unwrap();
assert!(result["properties"]["config"]["properties"]["key"].is_object());
}
#[test]
fn convert_mcp_schema_with_array_items() {
let schema = json!({
"type": "object",
"properties": {
"items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": { "type": "string" }
},
"additionalProperties": false
}
}
}
});
let result = convert_mcp_schema_to_gemini(Some(&schema)).unwrap();
let items_schema = &result["properties"]["items"]["items"];
assert!(items_schema.get("additionalProperties").is_none());
}
}
+20 -127
View File
@@ -1,5 +1,5 @@
//! Gemini runtime — drives Google Gemini API sessions as agent backends.
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use reqwest::Client;
@@ -10,7 +10,10 @@ use crate::agent_log::AgentLogWriter;
use crate::http::mcp::dispatch::dispatch_tool_call;
use crate::slog;
use super::super::{AgentEvent, TokenUsage};
use super::super::AgentEvent;
use super::api_common::{
CancellationFlag, check_loop_guard, done_result, extract_model, start_conversation_loop,
};
use super::{AgentRuntime, RuntimeContext, RuntimeResult, RuntimeStatus};
mod api;
@@ -40,14 +43,14 @@ struct GeminiFunctionCall {
/// 6. Tracks token usage from the API response metadata.
pub struct GeminiRuntime {
/// Whether a stop has been requested.
cancelled: Arc<AtomicBool>,
cancelled: CancellationFlag,
}
impl GeminiRuntime {
/// Create a new Gemini runtime instance.
pub fn new() -> Self {
Self {
cancelled: Arc::new(AtomicBool::new(false)),
cancelled: CancellationFlag::new(),
}
}
}
@@ -66,19 +69,7 @@ impl AgentRuntime for GeminiRuntime {
.to_string()
})?;
let model = if ctx.command.starts_with("gemini") {
// The pool puts the model into `command` for non-CLI runtimes,
// but also check args for a --model flag.
ctx.command.clone()
} else {
// Fall back to args: look for --model <value>
ctx.args
.iter()
.position(|a| a == "--model")
.and_then(|i| ctx.args.get(i + 1))
.cloned()
.unwrap_or_else(|| "gemini-2.5-pro".to_string())
};
let model = extract_model(&ctx, |c| c.starts_with("gemini"), "gemini-2.5-pro");
let app_ctx = ctx
.app_ctx
@@ -86,7 +77,7 @@ impl AgentRuntime for GeminiRuntime {
.ok_or_else(|| "Gemini runtime requires app_ctx to be set".to_string())?;
let client = Client::new();
let cancelled = Arc::clone(&self.cancelled);
let cancelled = self.cancelled.handle();
// Step 1: Load MCP tool definitions and convert to Gemini format.
let gemini_tools = convert_mcp_tools_to_gemini();
@@ -98,65 +89,14 @@ impl AgentRuntime for GeminiRuntime {
"parts": [{ "text": ctx.prompt }]
})];
let mut total_usage = TokenUsage {
input_tokens: 0,
output_tokens: 0,
cache_creation_input_tokens: 0,
cache_read_input_tokens: 0,
total_cost_usd: 0.0,
};
let emit = |event: AgentEvent| {
super::super::pty::emit_event(
event,
&tx,
&event_log,
log_writer.as_ref().map(|w| w.as_ref()),
);
};
emit(AgentEvent::Status {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
status: "running".to_string(),
});
let (emit, mut total_usage, mut turn) =
start_conversation_loop(&ctx, tx, event_log, log_writer);
// Step 3: Conversation loop.
let mut turn = 0u32;
let max_turns = 200; // Safety limit
loop {
if cancelled.load(Ordering::Relaxed) {
emit(AgentEvent::Error {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
message: "Agent was stopped by user".to_string(),
});
return Ok(RuntimeResult {
session_id: None,
token_usage: Some(total_usage),
exit_ok: true,
aborted_signal: false,
rate_limit_exit: false,
rate_limit_reset_at: None,
});
}
turn += 1;
if turn > max_turns {
emit(AgentEvent::Error {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
message: format!("Exceeded maximum turns ({max_turns})"),
});
return Ok(RuntimeResult {
session_id: None,
token_usage: Some(total_usage),
exit_ok: true,
aborted_signal: false,
rate_limit_exit: false,
rate_limit_reset_at: None,
});
if let Some(result) = check_loop_guard(&ctx, &cancelled, &mut turn, &total_usage, &emit)
{
return Ok(result);
}
slog!(
@@ -248,19 +188,7 @@ impl AgentRuntime for GeminiRuntime {
// If no function calls, the model is done.
if function_calls.is_empty() {
emit(AgentEvent::Done {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
session_id: None,
});
return Ok(RuntimeResult {
session_id: None,
token_usage: Some(total_usage),
exit_ok: true,
aborted_signal: false,
rate_limit_exit: false,
rate_limit_reset_at: None,
});
return Ok(done_result(&ctx, &emit, total_usage));
}
// Add the model's response to the conversation.
@@ -333,32 +261,15 @@ impl AgentRuntime for GeminiRuntime {
}
}
emit(AgentEvent::Done {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
session_id: None,
});
Ok(RuntimeResult {
session_id: None,
token_usage: Some(total_usage),
exit_ok: true,
aborted_signal: false,
rate_limit_exit: false,
rate_limit_reset_at: None,
})
Ok(done_result(&ctx, &emit, total_usage))
}
fn stop(&self) {
self.cancelled.store(true, Ordering::Relaxed);
self.cancelled.stop();
}
fn get_status(&self) -> RuntimeStatus {
if self.cancelled.load(Ordering::Relaxed) {
RuntimeStatus::Failed
} else {
RuntimeStatus::Idle
}
self.cancelled.status()
}
}
@@ -366,13 +277,8 @@ impl AgentRuntime for GeminiRuntime {
#[cfg(test)]
mod tests {
use super::super::api_common::test_runtime_context;
use super::*;
use crate::http::context::AppContext;
fn test_app_ctx() -> Arc<AppContext> {
let tmp = tempfile::tempdir().unwrap();
Arc::new(AppContext::new_test(tmp.path().to_path_buf()))
}
#[test]
fn gemini_runtime_stop_sets_cancelled() {
@@ -385,20 +291,7 @@ mod tests {
#[test]
fn model_extraction_from_command() {
// When command starts with "gemini", use it as model name
let ctx = RuntimeContext {
story_id: "1".to_string(),
agent_name: "coder".to_string(),
command: "gemini-2.5-pro".to_string(),
args: vec![],
prompt: "test".to_string(),
cwd: "/tmp".to_string(),
inactivity_timeout_secs: 300,
app_ctx: Some(test_app_ctx()),
session_id_to_resume: None,
fresh_prompt: None,
project_root: std::path::PathBuf::from("/tmp/project"),
model: None,
};
let ctx = test_runtime_context("gemini-2.5-pro", vec![]);
// The model extraction logic is inside start(), but we test the
// condition here.
+2 -6
View File
@@ -1,4 +1,5 @@
//! Agent runtimes — pluggable backends (Claude Code, Gemini, OpenAI) for running agents.
mod api_common;
mod claude_code;
mod gemini;
mod openai;
@@ -132,12 +133,7 @@ pub trait AgentRuntime: Send + Sync {
#[cfg(test)]
mod tests {
use super::*;
use crate::http::context::AppContext;
fn test_app_ctx() -> Arc<AppContext> {
let tmp = tempfile::tempdir().unwrap();
Arc::new(AppContext::new_test(tmp.path().to_path_buf()))
}
use api_common::test_app_ctx;
#[test]
fn runtime_context_fields() {
+25 -264
View File
@@ -1,5 +1,5 @@
//! OpenAI Codex runtime — drives OpenAI API sessions as agent backends.
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::atomic::Ordering;
use std::sync::{Arc, Mutex};
use reqwest::Client;
@@ -12,6 +12,10 @@ use crate::http::mcp::tools_list::list_tools;
use crate::slog;
use super::super::{AgentEvent, TokenUsage};
use super::api_common::{
CancellationFlag, build_system_text, check_loop_guard, clean_schema_properties, done_result,
extract_model, start_conversation_loop,
};
use super::{AgentRuntime, RuntimeContext, RuntimeResult, RuntimeStatus};
// ── Public runtime struct ────────────────────────────────────────────
@@ -28,14 +32,14 @@ use super::{AgentRuntime, RuntimeContext, RuntimeResult, RuntimeStatus};
/// 6. Tracks token usage from the API response.
pub struct OpenAiRuntime {
/// Whether a stop has been requested.
cancelled: Arc<AtomicBool>,
cancelled: CancellationFlag,
}
impl OpenAiRuntime {
/// Create a new OpenAI runtime instance.
pub fn new() -> Self {
Self {
cancelled: Arc::new(AtomicBool::new(false)),
cancelled: CancellationFlag::new(),
}
}
}
@@ -54,18 +58,11 @@ impl AgentRuntime for OpenAiRuntime {
.to_string()
})?;
let model = if ctx.command.starts_with("gpt") || ctx.command.starts_with("o") {
// The pool puts the model into `command` for non-CLI runtimes.
ctx.command.clone()
} else {
// Fall back to args: look for --model <value>
ctx.args
.iter()
.position(|a| a == "--model")
.and_then(|i| ctx.args.get(i + 1))
.cloned()
.unwrap_or_else(|| "gpt-4o".to_string())
};
let model = extract_model(
&ctx,
|c| c.starts_with("gpt") || c.starts_with("o"),
"gpt-4o",
);
let app_ctx = ctx
.app_ctx
@@ -73,7 +70,7 @@ impl AgentRuntime for OpenAiRuntime {
.ok_or_else(|| "OpenAI runtime requires app_ctx to be set".to_string())?;
let client = Client::new();
let cancelled = Arc::clone(&self.cancelled);
let cancelled = self.cancelled.handle();
// Step 1: Fetch MCP tool definitions and convert to OpenAI format.
let openai_tools = convert_mcp_tools_to_openai();
@@ -85,65 +82,14 @@ impl AgentRuntime for OpenAiRuntime {
json!({ "role": "user", "content": ctx.prompt }),
];
let mut total_usage = TokenUsage {
input_tokens: 0,
output_tokens: 0,
cache_creation_input_tokens: 0,
cache_read_input_tokens: 0,
total_cost_usd: 0.0,
};
let emit = |event: AgentEvent| {
super::super::pty::emit_event(
event,
&tx,
&event_log,
log_writer.as_ref().map(|w| w.as_ref()),
);
};
emit(AgentEvent::Status {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
status: "running".to_string(),
});
let (emit, mut total_usage, mut turn) =
start_conversation_loop(&ctx, tx, event_log, log_writer);
// Step 3: Conversation loop.
let mut turn = 0u32;
let max_turns = 200; // Safety limit
loop {
if cancelled.load(Ordering::Relaxed) {
emit(AgentEvent::Error {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
message: "Agent was stopped by user".to_string(),
});
return Ok(RuntimeResult {
session_id: None,
token_usage: Some(total_usage),
exit_ok: true,
aborted_signal: false,
rate_limit_exit: false,
rate_limit_reset_at: None,
});
}
turn += 1;
if turn > max_turns {
emit(AgentEvent::Error {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
message: format!("Exceeded maximum turns ({max_turns})"),
});
return Ok(RuntimeResult {
session_id: None,
token_usage: Some(total_usage),
exit_ok: true,
aborted_signal: false,
rate_limit_exit: false,
rate_limit_reset_at: None,
});
if let Some(result) = check_loop_guard(&ctx, &cancelled, &mut turn, &total_usage, &emit)
{
return Ok(result);
}
slog!(
@@ -218,19 +164,7 @@ impl AgentRuntime for OpenAiRuntime {
if tool_calls.is_none() || tool_calls.is_some_and(|tc| tc.is_empty()) {
// No tool calls — model is done.
emit(AgentEvent::Done {
story_id: ctx.story_id.clone(),
agent_name: ctx.agent_name.clone(),
session_id: None,
});
return Ok(RuntimeResult {
session_id: None,
token_usage: Some(total_usage),
exit_ok: true,
aborted_signal: false,
rate_limit_exit: false,
rate_limit_reset_at: None,
});
return Ok(done_result(&ctx, &emit, total_usage));
}
let tool_calls = tool_calls.unwrap();
@@ -297,38 +231,16 @@ impl AgentRuntime for OpenAiRuntime {
}
fn stop(&self) {
self.cancelled.store(true, Ordering::Relaxed);
self.cancelled.stop();
}
fn get_status(&self) -> RuntimeStatus {
if self.cancelled.load(Ordering::Relaxed) {
RuntimeStatus::Failed
} else {
RuntimeStatus::Idle
}
self.cancelled.status()
}
}
// ── Helper functions ─────────────────────────────────────────────────
/// Build the system message text from the RuntimeContext.
fn build_system_text(ctx: &RuntimeContext) -> String {
ctx.args
.iter()
.position(|a| a == "--append-system-prompt")
.and_then(|i| ctx.args.get(i + 1))
.cloned()
.unwrap_or_else(|| {
format!(
"You are an AI coding agent working on story {}. \
You have access to tools via function calling. \
Use them to complete the task. \
Work in the directory: {}",
ctx.story_id, ctx.cwd
)
})
}
/// Load MCP tool definitions directly and convert to OpenAI function-calling format.
fn convert_mcp_tools_to_openai() -> Vec<Value> {
let tools = list_tools();
@@ -377,7 +289,7 @@ fn convert_mcp_schema_to_openai(schema: Option<&Value>) -> Option<Value> {
});
if let Some(properties) = schema.get("properties") {
result["properties"] = clean_schema_properties(properties);
result["properties"] = clean_schema_properties(properties, false);
} else {
result["properties"] = json!({});
}
@@ -392,40 +304,6 @@ fn convert_mcp_schema_to_openai(schema: Option<&Value>) -> Option<Value> {
Some(result)
}
/// Recursively clean schema properties, removing unsupported keywords.
fn clean_schema_properties(properties: &Value) -> Value {
let Some(obj) = properties.as_object() else {
return properties.clone();
};
let mut cleaned = serde_json::Map::new();
for (key, value) in obj {
let mut prop = value.clone();
if let Some(p) = prop.as_object_mut() {
p.remove("$schema");
// Recursively clean nested object properties.
if let Some(nested_props) = p.get("properties").cloned() {
p.insert(
"properties".to_string(),
clean_schema_properties(&nested_props),
);
}
// Clean items schema for arrays.
if let Some(items) = p.get("items").cloned()
&& let Some(items_obj) = items.as_object()
{
let mut cleaned_items = items_obj.clone();
cleaned_items.remove("$schema");
p.insert("items".to_string(), Value::Object(cleaned_items));
}
}
cleaned.insert(key.clone(), prop);
}
Value::Object(cleaned)
}
/// Parse token usage from an OpenAI API response.
fn parse_usage(response: &Value) -> Option<TokenUsage> {
let usage = response.get("usage")?;
@@ -449,13 +327,8 @@ fn parse_usage(response: &Value) -> Option<TokenUsage> {
#[cfg(test)]
mod tests {
use super::super::api_common::test_runtime_context;
use super::*;
use crate::http::context::AppContext;
fn test_app_ctx() -> Arc<AppContext> {
let tmp = tempfile::tempdir().unwrap();
Arc::new(AppContext::new_test(tmp.path().to_path_buf()))
}
#[test]
fn convert_mcp_schema_simple_object() {
@@ -512,92 +385,6 @@ mod tests {
assert_eq!(name_prop["type"], "string");
}
#[test]
fn convert_mcp_schema_with_nested_objects() {
let schema = json!({
"type": "object",
"properties": {
"config": {
"type": "object",
"properties": {
"key": { "type": "string" }
}
}
}
});
let result = convert_mcp_schema_to_openai(Some(&schema)).unwrap();
assert!(result["properties"]["config"]["properties"]["key"].is_object());
}
#[test]
fn convert_mcp_schema_with_array_items() {
let schema = json!({
"type": "object",
"properties": {
"items": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": { "type": "string" }
},
"$schema": "http://json-schema.org/draft-07/schema#"
}
}
}
});
let result = convert_mcp_schema_to_openai(Some(&schema)).unwrap();
let items_schema = &result["properties"]["items"]["items"];
assert!(items_schema.get("$schema").is_none());
}
#[test]
fn build_system_text_uses_args() {
let ctx = RuntimeContext {
story_id: "42_story_test".to_string(),
agent_name: "coder-1".to_string(),
command: "gpt-4o".to_string(),
args: vec![
"--append-system-prompt".to_string(),
"Custom system prompt".to_string(),
],
prompt: "Do the thing".to_string(),
cwd: "/tmp/wt".to_string(),
inactivity_timeout_secs: 300,
app_ctx: Some(test_app_ctx()),
session_id_to_resume: None,
fresh_prompt: None,
project_root: std::path::PathBuf::from("/tmp/project"),
model: None,
};
assert_eq!(build_system_text(&ctx), "Custom system prompt");
}
#[test]
fn build_system_text_default() {
let ctx = RuntimeContext {
story_id: "42_story_test".to_string(),
agent_name: "coder-1".to_string(),
command: "gpt-4o".to_string(),
args: vec![],
prompt: "Do the thing".to_string(),
cwd: "/tmp/wt".to_string(),
inactivity_timeout_secs: 300,
app_ctx: Some(test_app_ctx()),
session_id_to_resume: None,
fresh_prompt: None,
project_root: std::path::PathBuf::from("/tmp/project"),
model: None,
};
let text = build_system_text(&ctx);
assert!(text.contains("42_story_test"));
assert!(text.contains("/tmp/wt"));
}
#[test]
fn parse_usage_valid() {
let response = json!({
@@ -631,39 +418,13 @@ mod tests {
#[test]
fn model_extraction_from_command_gpt() {
let ctx = RuntimeContext {
story_id: "1".to_string(),
agent_name: "coder".to_string(),
command: "gpt-4o".to_string(),
args: vec![],
prompt: "test".to_string(),
cwd: "/tmp".to_string(),
inactivity_timeout_secs: 300,
app_ctx: Some(test_app_ctx()),
session_id_to_resume: None,
fresh_prompt: None,
project_root: std::path::PathBuf::from("/tmp/project"),
model: None,
};
let ctx = test_runtime_context("gpt-4o", vec![]);
assert!(ctx.command.starts_with("gpt"));
}
#[test]
fn model_extraction_from_command_o3() {
let ctx = RuntimeContext {
story_id: "1".to_string(),
agent_name: "coder".to_string(),
command: "o3".to_string(),
args: vec![],
prompt: "test".to_string(),
cwd: "/tmp".to_string(),
inactivity_timeout_secs: 300,
app_ctx: Some(test_app_ctx()),
session_id_to_resume: None,
fresh_prompt: None,
project_root: std::path::PathBuf::from("/tmp/project"),
model: None,
};
let ctx = test_runtime_context("o3", vec![]);
assert!(ctx.command.starts_with("o"));
}
}
+250
View File
@@ -0,0 +1,250 @@
//! Transport-agnostic LLM turn driver shared by every chat transport that
//! forwards a non-command message to Claude Code (currently Discord and
//! WhatsApp). Formatting, chunking, and error-message differences between
//! transports are captured by the small [`TransportFormat`] trait; the
//! streaming/session/history plumbing lives once in [`run_llm_turn`].
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex as TokioMutex;
use crate::chat::ChatTransport;
use crate::chat::transport::matrix::{ConversationEntry, ConversationRole, RoomConversation};
use crate::http::context::PermissionDecision;
use crate::services::Services;
use crate::slog;
/// Per-transport formatting hooks needed to run a shared LLM turn.
pub trait TransportFormat: Send + Sync {
/// Convert markdown into the transport's native formatting and split it
/// into messages that respect the transport's size limit. Transports
/// without a size limit (or that don't chunk) return a single-element
/// vector.
fn format_and_chunk(&self, markdown: &str) -> Vec<String>;
/// Prefix used in `slog!` calls for this transport, e.g. `"[discord]"`.
fn log_prefix(&self) -> &'static str;
/// Convert an LLM error into a user-facing message.
fn format_error(&self, err: &str) -> String;
}
/// Build the prompt for an LLM turn, prepending any pending CRDT
/// pipeline-transition events as a `<system-reminder>` block.
pub fn build_llm_prompt(persona: &str, bot_name: &str, user: &str, user_message: &str) -> String {
let event_ctx = crate::llm_session::assemble_prompt_context(persona);
format!(
"{event_ctx}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{user}: {user_message}"
)
}
/// Everything a shared LLM turn needs beyond formatting: transport handle,
/// routing/history key, display name, and shared services/history state.
pub struct LlmTurnContext<'a> {
/// Chat transport used to send/post messages for this turn.
pub transport: Arc<dyn ChatTransport>,
/// Routing and history key (Discord channel id, WhatsApp phone number).
pub key: &'a str,
/// Display name stored in conversation history and used in the prompt.
pub user: &'a str,
/// Shared services bundle (project root, permissions, agents).
pub services: &'a Arc<Services>,
/// Per-key conversation history.
pub history: &'a Arc<TokioMutex<HashMap<String, RoomConversation>>>,
/// Maximum number of conversation entries to keep per key.
pub history_size: usize,
/// Persists conversation history to disk.
pub save_history: fn(&std::path::Path, &HashMap<String, RoomConversation>),
}
/// Forward a message to Claude Code and stream the response back through the
/// transport, handling permission prompts and conversation history.
///
/// `F` is taken by value (and cloned into the posting task) rather than by
/// reference because the posting task is a `tokio::spawn`ed future, which
/// requires `'static` — the formatter types are zero-sized markers, so
/// cloning is free.
pub async fn run_llm_turn<F: TransportFormat + Clone + 'static>(
ctx: LlmTurnContext<'_>,
format: F,
user_message: &str,
) {
use crate::chat::util::drain_complete_paragraphs;
use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult};
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::sync::watch;
let prefix = format.log_prefix();
// Look up existing session ID for this key.
let resume_session_id: Option<String> = {
let guard = ctx.history.lock().await;
guard.get(ctx.key).and_then(|conv| conv.session_id.clone())
};
let bot_name = &ctx.services.bot_name;
let persona = bot_name.to_lowercase();
let prompt = build_llm_prompt(&persona, bot_name, ctx.user, user_message);
let provider = ClaudeCodeProvider::new();
let (_cancel_tx, mut cancel_rx) = watch::channel(false);
// Channel for sending complete chunks to the posting task.
let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
let msg_tx_for_callback = msg_tx.clone();
// Spawn a task to post messages as they arrive.
let post_transport = Arc::clone(&ctx.transport);
let post_key = ctx.key.to_string();
let post_format = format.clone();
let post_task = tokio::spawn(async move {
while let Some(chunk) = msg_rx.recv().await {
for part in post_format.format_and_chunk(&chunk) {
let _ = post_transport.send_message(&post_key, &part, "").await;
}
}
});
// Shared buffer between the sync token callback and the async scope.
let buffer = Arc::new(std::sync::Mutex::new(String::new()));
let buffer_for_callback = Arc::clone(&buffer);
let sent_any_chunk = Arc::new(AtomicBool::new(false));
let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk);
let project_root_str = ctx.services.project_root.to_string_lossy().to_string();
let chat_fut = provider.chat_stream(
&prompt,
&project_root_str,
resume_session_id.as_deref(),
None,
None,
&mut cancel_rx,
move |token| {
let mut buf = buffer_for_callback.lock().unwrap();
buf.push_str(token);
let paragraphs = drain_complete_paragraphs(&mut buf);
for chunk in paragraphs {
sent_any_chunk_for_callback.store(true, Ordering::Relaxed);
let _ = msg_tx_for_callback.send(chunk);
}
},
|_thinking| {},
|_activity| {},
);
tokio::pin!(chat_fut);
// Register as a permission responder for the duration of this chat turn.
let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register();
let result = loop {
tokio::select! {
r = &mut chat_fut => break r,
Some(perm_fwd) = perm_rx.recv() => {
let prompt_msg = format!(
"**Permission Request**\n\nTool: `{}`\n```json\n{}\n```\n\nReply **yes** to approve or **no** to deny.",
perm_fwd.tool_name,
serde_json::to_string_pretty(&perm_fwd.tool_input)
.unwrap_or_else(|_| perm_fwd.tool_input.to_string()),
);
for part in format.format_and_chunk(&prompt_msg) {
let _ = ctx.transport.send_message(ctx.key, &part, "").await;
}
// Keyed by request_id (not just key) so a second
// concurrent request doesn't drop the first's sender.
ctx.services
.pending_perm_replies
.insert(ctx.key.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx)
.await;
// Spawn a timeout task: auto-deny if the user does not respond.
let pending = Arc::clone(&ctx.services.pending_perm_replies);
let timeout_key = ctx.key.to_string();
let timeout_request_id = perm_fwd.request_id.clone();
let timeout_transport = Arc::clone(&ctx.transport);
let timeout_secs = ctx.services.permission_timeout_secs;
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await;
if let Some(tx) = pending.remove_by_request_id(&timeout_key, &timeout_request_id).await {
let _ = tx.send(PermissionDecision::Deny);
let msg = "Permission request timed out — denied (fail-closed).";
let _ = timeout_transport.send_message(&timeout_key, msg, "").await;
}
});
}
}
};
// Flush remaining text.
let remaining = buffer.lock().unwrap().trim().to_string();
let did_send_any = sent_any_chunk.load(Ordering::Relaxed);
let (assistant_reply, new_session_id) = match result {
Ok(ClaudeCodeResult {
messages,
session_id,
..
}) => {
let reply = if !remaining.is_empty() {
let _ = msg_tx.send(remaining.clone());
remaining
} else if !did_send_any {
let last_text = messages
.iter()
.rev()
.find(|m| m.role == crate::llm::types::Role::Assistant && !m.content.is_empty())
.map(|m| m.content.clone())
.unwrap_or_default();
if !last_text.is_empty() {
let _ = msg_tx.send(last_text.clone());
}
last_text
} else {
remaining
};
slog!("{prefix} session_id from chat_stream: {:?}", session_id);
(reply, session_id)
}
Err(e) => {
slog!("{prefix} LLM error: {e}");
let err_msg = format.format_error(&e);
let _ = msg_tx.send(err_msg.clone());
(err_msg, None)
}
};
// Signal the posting task to finish and wait for it.
drop(msg_tx);
let _ = post_task.await;
// Record this exchange in conversation history.
if !assistant_reply.starts_with("Error processing") {
let mut guard = ctx.history.lock().await;
let conv = guard.entry(ctx.key.to_string()).or_default();
if new_session_id.is_some() {
conv.session_id = new_session_id;
}
conv.entries.push(ConversationEntry {
role: ConversationRole::User,
sender: ctx.user.to_string(),
content: user_message.to_string(),
});
conv.entries.push(ConversationEntry {
role: ConversationRole::Assistant,
sender: String::new(),
content: assistant_reply,
});
// Trim to configured maximum.
if conv.entries.len() > ctx.history_size {
let excess = conv.entries.len() - ctx.history_size;
conv.entries.drain(..excess);
}
(ctx.save_history)(&ctx.services.project_root, &guard);
}
}
+2
View File
@@ -12,6 +12,8 @@ pub mod compact;
pub mod dispatcher;
/// Chat history utilities — loading and serialising conversation history.
pub mod history;
/// Transport-agnostic LLM turn driver shared by Discord, WhatsApp, etc.
pub mod llm_turn;
pub(crate) mod lookup;
#[cfg(test)]
pub(crate) mod test_helpers;
+40 -187
View File
@@ -4,7 +4,9 @@ use std::collections::HashSet;
use std::sync::Arc;
use crate::chat::ChatTransport;
use crate::chat::transport::matrix::{ConversationEntry, ConversationRole, RoomConversation};
use crate::chat::transport::matrix::RoomConversation;
#[cfg(test)]
use crate::chat::transport::matrix::{ConversationEntry, ConversationRole};
use crate::chat::util::is_permission_approval;
use crate::http::context::PermissionDecision;
use crate::services::Services;
@@ -304,196 +306,43 @@ pub(super) async fn handle_incoming_message(
handle_llm_message(ctx, channel, user, message).await;
}
/// Build the prompt for a Discord LLM turn, prepending any pending
/// CRDT pipeline-transition events as a `<system-reminder>` block.
fn build_discord_llm_prompt(
persona: &str,
bot_name: &str,
user: &str,
user_message: &str,
) -> String {
let event_ctx = crate::llm_session::assemble_prompt_context(persona);
format!(
"{event_ctx}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{user}: {user_message}"
)
/// [`crate::chat::llm_turn::TransportFormat`] implementation for Discord:
/// single-message replies (no chunking) and no OAuth-link error handling.
#[derive(Clone, Copy)]
struct DiscordFormat;
impl crate::chat::llm_turn::TransportFormat for DiscordFormat {
fn format_and_chunk(&self, markdown: &str) -> Vec<String> {
vec![markdown_to_discord(markdown)]
}
fn log_prefix(&self) -> &'static str {
"[discord]"
}
fn format_error(&self, err: &str) -> String {
format!("Error processing your request: {err}")
}
}
/// Forward a message to Claude Code and send the response back via Discord.
async fn handle_llm_message(ctx: &DiscordContext, channel: &str, user: &str, user_message: &str) {
use crate::chat::util::drain_complete_paragraphs;
use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult};
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::sync::watch;
use crate::chat::llm_turn::{LlmTurnContext, run_llm_turn};
// Look up existing session ID for this channel.
let resume_session_id: Option<String> = {
let guard = ctx.history.lock().await;
guard.get(channel).and_then(|conv| conv.session_id.clone())
};
let bot_name = &ctx.services.bot_name;
let persona = bot_name.to_lowercase();
let prompt = build_discord_llm_prompt(&persona, bot_name, user, user_message);
let provider = ClaudeCodeProvider::new();
let (_cancel_tx, mut cancel_rx) = watch::channel(false);
// Channel for sending complete chunks to the Discord posting task.
let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
let msg_tx_for_callback = msg_tx.clone();
// Spawn a task to post messages as they arrive.
let post_transport = Arc::clone(&ctx.transport) as Arc<dyn ChatTransport>;
let post_channel = channel.to_string();
let post_task = tokio::spawn(async move {
while let Some(chunk) = msg_rx.recv().await {
let formatted = markdown_to_discord(&chunk);
let _ = post_transport
.send_message(&post_channel, &formatted, "")
.await;
}
});
// Shared buffer between the sync token callback and the async scope.
let buffer = Arc::new(std::sync::Mutex::new(String::new()));
let buffer_for_callback = Arc::clone(&buffer);
let sent_any_chunk = Arc::new(AtomicBool::new(false));
let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk);
let project_root_str = ctx.services.project_root.to_string_lossy().to_string();
let chat_fut = provider.chat_stream(
&prompt,
&project_root_str,
resume_session_id.as_deref(),
None,
None,
&mut cancel_rx,
move |token| {
let mut buf = buffer_for_callback.lock().unwrap();
buf.push_str(token);
let paragraphs = drain_complete_paragraphs(&mut buf);
for chunk in paragraphs {
sent_any_chunk_for_callback.store(true, Ordering::Relaxed);
let _ = msg_tx_for_callback.send(chunk);
}
run_llm_turn(
LlmTurnContext {
transport: Arc::clone(&ctx.transport) as Arc<dyn ChatTransport>,
key: channel,
user,
services: &ctx.services,
history: &ctx.history,
history_size: ctx.history_size,
save_history: save_discord_history,
},
|_thinking| {},
|_activity| {},
);
tokio::pin!(chat_fut);
// Register as a permission responder for the duration of this chat turn.
let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register();
let result = loop {
tokio::select! {
r = &mut chat_fut => break r,
Some(perm_fwd) = perm_rx.recv() => {
let prompt_msg = format!(
"**Permission Request**\n\nTool: `{}`\n```json\n{}\n```\n\nReply **yes** to approve or **no** to deny.",
perm_fwd.tool_name,
serde_json::to_string_pretty(&perm_fwd.tool_input)
.unwrap_or_else(|_| perm_fwd.tool_input.to_string()),
);
let formatted = markdown_to_discord(&prompt_msg);
let _ = ctx.transport.send_message(channel, &formatted, "").await;
// Keyed by request_id (not just channel) so a second
// concurrent request doesn't drop the first's sender.
ctx.services
.pending_perm_replies
.insert(channel.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx)
.await;
// Spawn a timeout task: auto-deny if the user does not respond.
let pending = Arc::clone(&ctx.services.pending_perm_replies);
let timeout_channel = channel.to_string();
let timeout_request_id = perm_fwd.request_id.clone();
let timeout_transport = Arc::clone(&ctx.transport) as Arc<dyn ChatTransport>;
let timeout_secs = ctx.services.permission_timeout_secs;
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await;
if let Some(tx) = pending.remove_by_request_id(&timeout_channel, &timeout_request_id).await {
let _ = tx.send(PermissionDecision::Deny);
let msg = "Permission request timed out — denied (fail-closed).";
let _ = timeout_transport.send_message(&timeout_channel, msg, "").await;
}
});
}
}
};
// Flush remaining text.
let remaining = buffer.lock().unwrap().trim().to_string();
let did_send_any = sent_any_chunk.load(Ordering::Relaxed);
let (assistant_reply, new_session_id) = match result {
Ok(ClaudeCodeResult {
messages,
session_id,
..
}) => {
let reply = if !remaining.is_empty() {
let _ = msg_tx.send(remaining.clone());
remaining
} else if !did_send_any {
let last_text = messages
.iter()
.rev()
.find(|m| m.role == crate::llm::types::Role::Assistant && !m.content.is_empty())
.map(|m| m.content.clone())
.unwrap_or_default();
if !last_text.is_empty() {
let _ = msg_tx.send(last_text.clone());
}
last_text
} else {
remaining
};
slog!("[discord] session_id from chat_stream: {:?}", session_id);
(reply, session_id)
}
Err(e) => {
slog!("[discord] LLM error: {e}");
let err_msg = format!("Error processing your request: {e}");
let _ = msg_tx.send(err_msg.clone());
(err_msg, None)
}
};
// Signal the posting task to finish and wait for it.
drop(msg_tx);
let _ = post_task.await;
// Record this exchange in conversation history.
if !assistant_reply.starts_with("Error processing") {
let mut guard = ctx.history.lock().await;
let conv = guard.entry(channel.to_string()).or_default();
if new_session_id.is_some() {
conv.session_id = new_session_id;
}
conv.entries.push(ConversationEntry {
role: ConversationRole::User,
sender: user.to_string(),
content: user_message.to_string(),
});
conv.entries.push(ConversationEntry {
role: ConversationRole::Assistant,
sender: String::new(),
content: assistant_reply,
});
// Trim to configured maximum.
if conv.entries.len() > ctx.history_size {
let excess = conv.entries.len() - ctx.history_size;
conv.entries.drain(..excess);
}
save_discord_history(&ctx.services.project_root, &guard);
}
DiscordFormat,
user_message,
)
.await;
}
// ── Tests ───────────────────────────────────────────────────────────────
@@ -635,8 +484,12 @@ mod tests {
at: chrono::Utc::now(),
});
let prompt =
build_discord_llm_prompt("discord-ch-test", "Timmy", "@alice", "what is the status?");
let prompt = crate::chat::llm_turn::build_llm_prompt(
"discord-ch-test",
"Timmy",
"@alice",
"what is the status?",
);
assert!(
prompt.contains("<system-reminder>"),
@@ -130,10 +130,6 @@ pub struct BotContext {
/// Minimum seconds between repeated `compact` suggestions for the same
/// room. From `bot.toml`'s `compact_suggest_cooldown_secs`.
pub compact_suggest_cooldown_secs: i64,
/// Seconds a turn may spend on tool calls before emitting any
/// user-facing text before the bot posts a "digging in" notice. From
/// `bot.toml`'s `digging_in_threshold_secs`.
pub digging_in_threshold_secs: u64,
}
impl BotContext {
@@ -365,7 +361,6 @@ mod tests {
compact_seed_max_bytes: 8_000,
cache_read_suggest_threshold: 50_000,
compact_suggest_cooldown_secs: 3_600,
digging_in_threshold_secs: 15,
}
}
@@ -9,7 +9,6 @@ use matrix_sdk::ruma::{OwnedEventId, OwnedRoomId};
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::Mutex as TokioMutex;
use tokio::sync::watch;
@@ -19,34 +18,10 @@ use super::super::history::{ConversationEntry, ConversationRole, save_history};
use super::format_user_prompt;
/// Text posted to the room by [`spawn_digging_in_watcher`] when a turn runs
/// long without emitting any user-facing text, and by `on_room_message` as
/// an immediate acknowledgement when a message is first received (story 1239).
/// Text posted by `on_room_message` as an immediate acknowledgement when a
/// message is first received (story 1239).
pub(super) const DIGGING_IN_MESSAGE: &str = "Working...";
/// Spawns a background watcher that posts a single "digging in" notice to
/// `room_id` if `threshold` elapses before `sent_any_text` becomes `true`.
///
/// Callers must abort the returned [`tokio::task::JoinHandle`] once the turn
/// completes so a turn that finishes under the threshold — with or without
/// text — never triggers the notice after the fact.
pub(in crate::chat::transport::matrix::bot) fn spawn_digging_in_watcher(
transport: Arc<dyn ChatTransport>,
room_id: String,
sent_any_text: Arc<AtomicBool>,
threshold: Duration,
) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
tokio::time::sleep(threshold).await;
if !sent_any_text.load(Ordering::Relaxed) {
let html = markdown_to_html(DIGGING_IN_MESSAGE);
let _ = transport
.send_message(&room_id, DIGGING_IN_MESSAGE, &html)
.await;
}
})
}
/// One live-progress update to apply to the room's placeholder message
/// while a turn runs (story 1240).
enum ProgressUpdate {
@@ -248,21 +223,10 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message(
);
tokio::pin!(chat_fut);
// Notify the room if this turn spends longer than the configured
// threshold on tool calls before emitting any user-facing text, so a
// long silent turn doesn't look hung. Aborted below once the turn ends.
let digging_in_task = spawn_digging_in_watcher(
Arc::clone(&ctx.transport),
room_id_str.clone(),
Arc::clone(&sent_any_chunk),
Duration::from_secs(ctx.digging_in_threshold_secs),
);
// Permission requests are handled by the persistent permission_listener
// task spawned at bot startup (story 884) — they no longer route through
// per-message handlers. Just await chat_fut.
let result = (&mut chat_fut).await;
digging_in_task.abort();
// Flush any remaining text that didn't end with a paragraph boundary.
let remaining = buffer.lock().unwrap().trim().to_string();
@@ -503,107 +467,6 @@ mod tests {
}
}
/// AC 1: a turn that runs longer than the threshold without sending any
/// text gets exactly one "Working..." notice.
#[tokio::test]
async fn digging_in_fires_after_threshold_when_no_text_sent() {
let transport = Arc::new(CapturingTransport::new());
let sent_any_text = Arc::new(AtomicBool::new(false));
let handle = spawn_digging_in_watcher(
transport.clone() as Arc<dyn ChatTransport>,
"!room:example.com".to_string(),
Arc::clone(&sent_any_text),
Duration::from_millis(30),
);
handle.await.unwrap();
assert_eq!(transport.sent_count(), 1);
assert_eq!(transport.last_message().unwrap(), "Working...");
}
/// AC 2: if text is sent before the threshold elapses, the watcher must
/// not post anything.
#[tokio::test]
async fn digging_in_does_not_fire_when_text_sent_before_threshold() {
let transport = Arc::new(CapturingTransport::new());
let sent_any_text = Arc::new(AtomicBool::new(false));
let handle = spawn_digging_in_watcher(
transport.clone() as Arc<dyn ChatTransport>,
"!room:example.com".to_string(),
Arc::clone(&sent_any_text),
Duration::from_millis(30),
);
sent_any_text.store(true, Ordering::Relaxed);
handle.await.unwrap();
assert_eq!(transport.sent_count(), 0);
}
/// AC 2: a fast turn that completes (and is aborted by its caller)
/// before the threshold elapses must not post anything, even if it
/// never sent any text either.
#[tokio::test]
async fn digging_in_does_not_fire_when_aborted_before_threshold() {
let transport = Arc::new(CapturingTransport::new());
let sent_any_text = Arc::new(AtomicBool::new(false));
let handle = spawn_digging_in_watcher(
transport.clone() as Arc<dyn ChatTransport>,
"!room:example.com".to_string(),
Arc::clone(&sent_any_text),
Duration::from_millis(200),
);
handle.abort();
tokio::time::sleep(Duration::from_millis(250)).await;
assert_eq!(transport.sent_count(), 0);
}
/// AC 3: the notice fires at most once per turn — even waiting well past
/// the threshold never produces a second message.
#[tokio::test]
async fn digging_in_fires_at_most_once_even_after_extra_wait() {
let transport = Arc::new(CapturingTransport::new());
let sent_any_text = Arc::new(AtomicBool::new(false));
let handle = spawn_digging_in_watcher(
transport.clone() as Arc<dyn ChatTransport>,
"!room:example.com".to_string(),
Arc::clone(&sent_any_text),
Duration::from_millis(20),
);
handle.await.unwrap();
tokio::time::sleep(Duration::from_millis(100)).await;
assert_eq!(transport.sent_count(), 1);
}
/// AC 5: state is fresh per call — spawning a second, independent
/// watcher (as a new turn would) does not inherit any state from a
/// prior watcher's `sent_any_text` flag.
#[tokio::test]
async fn digging_in_state_does_not_leak_across_turns() {
let transport = Arc::new(CapturingTransport::new());
// First "turn": text sent before threshold, so no notice.
let first_sent_any_text = Arc::new(AtomicBool::new(false));
let first_handle = spawn_digging_in_watcher(
transport.clone() as Arc<dyn ChatTransport>,
"!room:example.com".to_string(),
Arc::clone(&first_sent_any_text),
Duration::from_millis(30),
);
first_sent_any_text.store(true, Ordering::Relaxed);
first_handle.await.unwrap();
assert_eq!(transport.sent_count(), 0);
// Second "turn": fresh flag, no text sent — must fire independently
// of the first turn's outcome.
let second_sent_any_text = Arc::new(AtomicBool::new(false));
let second_handle = spawn_digging_in_watcher(
transport.clone() as Arc<dyn ChatTransport>,
"!room:example.com".to_string(),
Arc::clone(&second_sent_any_text),
Duration::from_millis(30),
);
second_handle.await.unwrap();
assert_eq!(transport.sent_count(), 1);
}
// ── run_progress_updates (story 1240) ─────────────────────────────────
fn spawn_progress(
@@ -924,7 +924,6 @@ pub(in crate::chat::transport::matrix::bot) async fn on_room_message(
rebuild_cmd.drain_timeout_secs,
rebuild_cmd.force,
store,
ctx.gateway_port,
&ctx.services.project_root,
)
.await
@@ -2092,7 +2091,6 @@ mod tests {
compact_seed_max_bytes: 8_000,
cache_read_suggest_threshold: 50_000,
compact_suggest_cooldown_secs: 3_600,
digging_in_threshold_secs: 15,
}
}
@@ -343,7 +343,6 @@ pub async fn run_bot(
compact_seed_max_bytes: config.compact_seed_max_bytes,
cache_read_suggest_threshold: config.cache_read_suggest_threshold,
compact_suggest_cooldown_secs: config.compact_suggest_cooldown_secs,
digging_in_threshold_secs: config.digging_in_threshold_secs,
};
slog!(
@@ -172,49 +172,6 @@ history_size = 50
assert_eq!(config.history_size, 50);
}
/// AC4: `digging_in_threshold_secs` defaults to 15 when unset in bot.toml.
#[test]
fn load_uses_default_digging_in_threshold_secs() {
let tmp = tempfile::tempdir().unwrap();
let sk = tmp.path().join(".huskies");
fs::create_dir_all(&sk).unwrap();
fs::write(
sk.join("bot.toml"),
r#"
homeserver = "https://matrix.example.com"
username = "@bot:example.com"
password = "secret"
room_ids = ["!abc:example.com"]
enabled = true
"#,
)
.unwrap();
let config = BotConfig::load(tmp.path()).unwrap();
assert_eq!(config.digging_in_threshold_secs, 15);
}
/// AC4: `digging_in_threshold_secs` can be overridden in bot.toml.
#[test]
fn load_respects_custom_digging_in_threshold_secs() {
let tmp = tempfile::tempdir().unwrap();
let sk = tmp.path().join(".huskies");
fs::create_dir_all(&sk).unwrap();
fs::write(
sk.join("bot.toml"),
r#"
homeserver = "https://matrix.example.com"
username = "@bot:example.com"
password = "secret"
room_ids = ["!abc:example.com"]
enabled = true
digging_in_threshold_secs = 30
"#,
)
.unwrap();
let config = BotConfig::load(tmp.path()).unwrap();
assert_eq!(config.digging_in_threshold_secs, 30);
}
#[test]
fn load_reads_display_name() {
let tmp = tempfile::tempdir().unwrap();
@@ -31,12 +31,6 @@ pub(super) fn default_compact_suggest_cooldown_secs() -> i64 {
3_600
}
/// Default threshold (seconds) a turn may spend on tool calls before
/// emitting any user-facing text before the bot posts a "digging in" notice.
pub(super) fn default_digging_in_threshold_secs() -> u64 {
15
}
pub(super) fn default_transport() -> String {
"matrix".to_string()
}
@@ -229,10 +223,4 @@ pub struct BotConfig {
/// room, so a busy room isn't spammed every turn. Defaults to 3600 (1h).
#[serde(default = "default_compact_suggest_cooldown_secs")]
pub compact_suggest_cooldown_secs: i64,
/// Seconds a turn may spend on tool calls before emitting any
/// user-facing text before the bot posts a "digging in" notice to the
/// room, so a silent long-running turn doesn't look hung. Defaults to 15.
#[serde(default = "default_digging_in_threshold_secs")]
pub digging_in_threshold_secs: u64,
}
@@ -959,7 +959,6 @@ mod tests {
compact_seed_max_bytes: 8_000,
cache_read_suggest_threshold: 50_000,
compact_suggest_cooldown_secs: 3_600,
digging_in_threshold_secs: 15,
}
}
}
@@ -20,7 +20,6 @@ use crate::service::gateway::io::save_config;
use std::collections::BTreeMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
/// Default seconds to wait for in-flight work to drain before refusing.
@@ -102,7 +101,6 @@ pub async fn handle_project_rebuild(
drain_timeout_secs: u64,
force: bool,
projects_store: &Arc<RwLock<BTreeMap<String, ProjectEntry>>>,
gateway_port: Option<u16>,
config_dir: &Path,
) -> String {
// ── 1. Validate project ──────────────────────────────────────────────────
@@ -142,21 +140,6 @@ pub async fn handle_project_rebuild(
);
}
// ── 1b. Capture the running binary's git_hash before we tear it down ──────
// Best-effort: an unreachable or pre-version-endpoint sled just yields `None`,
// in which case the post-rebuild reconciliation below can't compare old vs
// new and falls back to whatever the published fleet artifact says.
let http_client = reqwest::Client::builder()
.timeout(Duration::from_secs(15))
.build()
.unwrap_or_default();
let old_git_hash = match project_url.as_deref() {
Some(url) => super::sled_upgrade::fetch_sled_version(&http_client, url)
.await
.map(|(_version, hash)| hash),
None => None,
};
// ── 2. Drain check ───────────────────────────────────────────────────────
let container_name = format!("huskies-{name}");
if !force
@@ -317,20 +300,6 @@ pub async fn handle_project_rebuild(
crate::slog!("[project-rebuild] Rebuilt '{name}': image={image_hash} container={container_id}");
// ── 7. Make sure the rebuilt sled isn't running an older binary than before ──
let artifact_source = super::sled_upgrade::resolve_artifact_source(gateway_port);
let version_note = reconcile_post_rebuild_version(
name,
old_git_hash.as_deref(),
&container_url,
artifact_source,
projects_store,
config_dir,
&http_client,
60,
)
.await;
format!(
"Project **{name}** rebuilt.\n\
- New image: `{image}` (`{image_short}`)\n\
@@ -338,103 +307,10 @@ pub async fn handle_project_rebuild(
- State: `pipeline.db` and CRDT preserved (same volume bind-mount)\n\
- Port: {port} (unchanged)\n\
\n\
Use `switch {name}` then `status` to verify the pipeline.{version_note}"
Use `switch {name}` then `status` to verify the pipeline."
)
}
/// Ensure the freshly-rebuilt sled at `container_url` isn't running an older
/// binary than the one it had before the rebuild (AC 13, story 1231).
///
/// Base images are only refreshed by a manual `script/build-project-images`
/// run, so `project-rebuild` can otherwise silently downgrade a sled that was
/// live-upgraded past whatever binary is baked into the image. This waits for
/// the new container to come up, compares its `git_hash` against the
/// gateway's published fleet artifact, and — when they differ — self-heals by
/// running the same in-place upgrade `upgrade <name>` uses. When there's no
/// published artifact to converge to, it falls back to comparing against the
/// pre-rebuild hash so a real downgrade is at least surfaced as a warning
/// instead of passing silently.
///
/// `artifact_source` is the already-resolved published-artifact lookup (see
/// [`super::sled_upgrade::resolve_artifact_source`]) — `Err` means no fleet
/// artifact has been published yet.
///
/// Returns a Markdown snippet (starting with `\n\n`) to append to the rebuild
/// reply, or an empty string when the sled already matches the current build.
#[allow(clippy::too_many_arguments)]
async fn reconcile_post_rebuild_version(
name: &str,
old_git_hash: Option<&str>,
container_url: &str,
artifact_source: Result<(String, Option<String>), String>,
projects_store: &Arc<RwLock<BTreeMap<String, ProjectEntry>>>,
config_dir: &Path,
client: &reqwest::Client,
health_timeout_secs: u64,
) -> String {
let health_url = format!("{}/health", container_url.trim_end_matches('/'));
if !super::sled_upgrade::wait_for_health(client, &health_url, health_timeout_secs).await {
return format!(
"\n\n⚠️ **Warning:** the rebuilt sled did not respond to `/health` within 60s — \
cannot verify its binary version. Check `docker logs huskies-{name}` and run \
`upgrade {name}` once it's reachable."
);
}
let Some((_new_version, new_hash)) =
super::sled_upgrade::fetch_sled_version(client, container_url).await
else {
return "\n\n⚠️ **Warning:** the rebuilt sled is healthy but `/api/version` is unavailable — \
cannot verify it isn't running a stale binary baked into the base image."
.to_string();
};
let (source_url, expected_hash) = match artifact_source {
Ok(v) => v,
Err(_) => {
// No published fleet artifact to converge to — the best we can do is
// flag an actual change from what was running before the rebuild.
return match old_git_hash {
Some(old) if old != new_hash => format!(
"\n\n⚠️ **Warning:** the rebuilt sled is now on `{new_hash}` (was `{old}`) and \
no fleet artifact is published to verify or fix this. Run `release` then \
`upgrade {name}`."
),
_ => String::new(),
};
}
};
if expected_hash.as_deref() == Some(new_hash.as_str()) {
return String::new(); // base image already bakes the current fleet build.
}
// The base image baked an older (or merely different) binary than the fleet
// is currently on — self-heal the same way `upgrade <name>` would.
let outcome = super::sled_upgrade::run_sled_upgrade(
name,
container_url,
&source_url,
expected_hash,
projects_store,
config_dir,
|_| async {},
)
.await;
if outcome.contains("matches published artifact") || outcome.starts_with("upgraded to v") {
format!(
"\n\n♻️ The base image baked an older binary (`{new_hash}`); self-healed in place — \
{outcome}"
)
} else {
format!(
"\n\n⚠️ **Warning:** the base image baked an older binary (`{new_hash}`) and the \
automatic self-heal failed: {outcome}\nRun `upgrade {name}` manually."
)
}
}
/// Wait for active Claude agent processes in the container to exit.
///
/// Polls every 5 seconds until no `claude` processes remain or `timeout_secs` elapses.
@@ -634,7 +510,7 @@ mod tests {
async fn rebuild_unknown_project_returns_error() {
let store = make_store(vec![]);
let dir = tempfile::tempdir().unwrap();
let result = handle_project_rebuild("nonexistent", 0, true, &store, None, dir.path()).await;
let result = handle_project_rebuild("nonexistent", 0, true, &store, dir.path()).await;
assert!(
result.contains("not found"),
"expected 'not found': {result}"
@@ -654,7 +530,7 @@ mod tests {
},
)]);
let dir = tempfile::tempdir().unwrap();
let result = handle_project_rebuild("myapp", 0, true, &store, None, dir.path()).await;
let result = handle_project_rebuild("myapp", 0, true, &store, dir.path()).await;
assert!(
result.contains("host_path"),
"expected 'host_path' mention: {result}"
@@ -674,7 +550,7 @@ mod tests {
},
)]);
let dir = tempfile::tempdir().unwrap();
let result = handle_project_rebuild("myapp", 0, true, &store, None, dir.path()).await;
let result = handle_project_rebuild("myapp", 0, true, &store, dir.path()).await;
assert!(
result.contains("does not exist"),
"expected 'does not exist': {result}"
@@ -710,8 +586,7 @@ mod tests {
)]);
let config_dir = tempfile::tempdir().unwrap();
let result =
handle_project_rebuild("myapp", 0, true, &store, None, config_dir.path()).await;
let result = handle_project_rebuild("myapp", 0, true, &store, config_dir.path()).await;
// (a) Step naming: one of several possible failure steps depending on what Docker
// binaries are available in the test environment, or a success reply.
@@ -732,181 +607,4 @@ mod tests {
"project 'myapp' must remain registered after failed rebuild: {result}"
);
}
// ── reconcile_post_rebuild_version (AC 13) ────────────────────────────
/// Spawn a minimal HTTP server that answers every `/health` request with
/// 200 and every `/api/version` request with `git_hash`. Serves
/// connections in a loop so both `wait_for_health` and
/// `fetch_sled_version` can hit it independently within one test.
fn spawn_version_server(git_hash: &str) -> (String, tokio::task::JoinHandle<()>) {
let git_hash = git_hash.to_string();
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let listener = tokio::net::TcpListener::from_std(listener).unwrap();
let port = listener.local_addr().unwrap().port();
let handle = tokio::spawn(async move {
loop {
let Ok((mut stream, _)) = listener.accept().await else {
break;
};
let git_hash = git_hash.clone();
tokio::spawn(async move {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let mut buf = [0u8; 4096];
let n = stream.read(&mut buf).await.unwrap_or(0);
let req = String::from_utf8_lossy(&buf[..n]);
let path = req.lines().next().unwrap_or("").to_string();
let body = if path.contains("/api/version") {
serde_json::json!({"version": "0.14.2", "git_hash": git_hash}).to_string()
} else {
"ok".to_string()
};
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{}",
body.len(),
body
);
let _ = stream.write_all(resp.as_bytes()).await;
});
}
});
(format!("http://127.0.0.1:{port}"), handle)
}
#[tokio::test]
async fn reconcile_empty_when_new_hash_matches_published_artifact() {
let (url, srv) = spawn_version_server("current-hash");
let store = make_store(vec![]);
let config_dir = tempfile::tempdir().unwrap();
let client = reqwest::Client::new();
let note = reconcile_post_rebuild_version(
"myapp",
Some("old-hash"),
&url,
Ok((
"http://unused".to_string(),
Some("current-hash".to_string()),
)),
&store,
config_dir.path(),
&client,
2,
)
.await;
assert_eq!(note, "", "already-current build must produce no note");
srv.abort();
}
#[tokio::test]
async fn reconcile_empty_when_no_artifact_and_hash_unchanged() {
let (url, srv) = spawn_version_server("same-hash");
let store = make_store(vec![]);
let config_dir = tempfile::tempdir().unwrap();
let client = reqwest::Client::new();
let note = reconcile_post_rebuild_version(
"myapp",
Some("same-hash"),
&url,
Err("no published artifact".to_string()),
&store,
config_dir.path(),
&client,
2,
)
.await;
assert_eq!(note, "", "unchanged hash must produce no note");
srv.abort();
}
#[tokio::test]
async fn reconcile_warns_when_no_artifact_and_hash_changed() {
let (url, srv) = spawn_version_server("stale-baked-hash");
let store = make_store(vec![]);
let config_dir = tempfile::tempdir().unwrap();
let client = reqwest::Client::new();
let note = reconcile_post_rebuild_version(
"myapp",
Some("newer-hash-that-was-running"),
&url,
Err("no published artifact".to_string()),
&store,
config_dir.path(),
&client,
2,
)
.await;
assert!(note.contains("Warning"), "expected a warning: {note}");
assert!(
note.contains("stale-baked-hash") && note.contains("newer-hash-that-was-running"),
"warning should name both hashes: {note}"
);
srv.abort();
}
#[tokio::test]
async fn reconcile_warns_when_health_probe_fails() {
let store = make_store(vec![]);
let config_dir = tempfile::tempdir().unwrap();
let client = reqwest::Client::builder()
.timeout(std::time::Duration::from_millis(200))
.build()
.unwrap();
let note = reconcile_post_rebuild_version(
"myapp",
Some("old-hash"),
"http://127.0.0.1:1", // nothing listens on port 1
Err("no published artifact".to_string()),
&store,
config_dir.path(),
&client,
2,
)
.await;
assert!(
note.contains("did not respond to `/health`"),
"expected a health-probe warning: {note}"
);
}
#[tokio::test]
async fn reconcile_attempts_self_heal_and_reports_failure_when_artifact_unreachable() {
let (url, srv) = spawn_version_server("stale-baked-hash");
let store = make_store(vec![]);
let config_dir = tempfile::tempdir().unwrap();
let client = reqwest::Client::new();
let note = reconcile_post_rebuild_version(
"myapp",
Some("stale-baked-hash"),
&url,
Ok((
"http://127.0.0.1:1/api/artifacts/huskies-linux-arm64".to_string(),
Some("current-fleet-hash".to_string()),
)),
&store,
config_dir.path(),
&client,
2,
)
.await;
assert!(
note.contains("self-heal failed"),
"mismatch should trigger a self-heal attempt that reports failure: {note}"
);
assert!(
note.contains("stale-baked-hash"),
"note should name the stale hash: {note}"
);
srv.abort();
}
}
@@ -102,7 +102,6 @@ mod tests {
compact_seed_max_bytes: 8_000,
cache_read_suggest_threshold: 50_000,
compact_suggest_cooldown_secs: 3_600,
digging_in_threshold_secs: 15,
};
run_projects_list(&ctx).await
}
@@ -225,7 +224,6 @@ mod tests {
compact_seed_max_bytes: 8_000,
cache_read_suggest_threshold: 50_000,
compact_suggest_cooldown_secs: 3_600,
digging_in_threshold_secs: 15,
};
let response = run_projects_list(&ctx).await;
assert!(
@@ -2,204 +2,56 @@
use std::sync::Arc;
use crate::chat::transport::matrix::{ConversationEntry, ConversationRole};
use crate::http::context::PermissionDecision;
use crate::slog;
use super::super::WhatsAppWebhookContext;
use super::super::format::{chunk_for_whatsapp, markdown_to_whatsapp};
use super::super::history::save_whatsapp_history;
/// [`crate::chat::llm_turn::TransportFormat`] implementation for WhatsApp:
/// messages are markdown-converted then split to fit the platform's size
/// limit, and LLM errors are checked for an OAuth login link.
#[derive(Clone, Copy)]
struct WhatsAppFormat;
impl crate::chat::llm_turn::TransportFormat for WhatsAppFormat {
fn format_and_chunk(&self, markdown: &str) -> Vec<String> {
chunk_for_whatsapp(&markdown_to_whatsapp(markdown))
}
fn log_prefix(&self) -> &'static str {
"[whatsapp]"
}
fn format_error(&self, err: &str) -> String {
if let Some(url) = crate::llm::oauth::extract_login_url_from_error(err) {
format!("Authentication required. Log in to Claude here: {url}")
} else {
format!("Error processing your request: {err}")
}
}
}
/// Forward a message to Claude Code and send the response back via WhatsApp.
pub(super) async fn handle_llm_message(
ctx: &WhatsAppWebhookContext,
sender: &str,
user_message: &str,
) {
use crate::chat::util::drain_complete_paragraphs;
use crate::llm::providers::claude_code::{ClaudeCodeProvider, ClaudeCodeResult};
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::sync::watch;
use crate::chat::llm_turn::{LlmTurnContext, run_llm_turn};
// Look up existing session ID for this sender.
let resume_session_id: Option<String> = {
let guard = ctx.history.lock().await;
guard.get(sender).and_then(|conv| conv.session_id.clone())
};
let bot_name = &ctx.services.bot_name;
let persona = bot_name.to_lowercase();
let event_ctx = crate::llm_session::assemble_prompt_context(&persona);
let prompt = format!(
"{event_ctx}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n\n{sender}: {user_message}"
);
let provider = ClaudeCodeProvider::new();
let (_cancel_tx, mut cancel_rx) = watch::channel(false);
// Channel for sending complete chunks to the WhatsApp posting task.
let (msg_tx, mut msg_rx) = tokio::sync::mpsc::unbounded_channel::<String>();
let msg_tx_for_callback = msg_tx.clone();
// Spawn a task to post messages as they arrive.
let post_transport = Arc::clone(&ctx.transport);
let post_sender = sender.to_string();
let post_task = tokio::spawn(async move {
while let Some(chunk) = msg_rx.recv().await {
// Convert Markdown to WhatsApp formatting, then split into sized chunks.
let formatted = markdown_to_whatsapp(&chunk);
for part in chunk_for_whatsapp(&formatted) {
let _ = post_transport.send_message(&post_sender, &part, "").await;
}
}
});
// Shared buffer between the sync token callback and the async scope.
let buffer = Arc::new(std::sync::Mutex::new(String::new()));
let buffer_for_callback = Arc::clone(&buffer);
let sent_any_chunk = Arc::new(AtomicBool::new(false));
let sent_any_chunk_for_callback = Arc::clone(&sent_any_chunk);
let project_root_str = ctx.services.project_root.to_string_lossy().to_string();
let chat_fut = provider.chat_stream(
&prompt,
&project_root_str,
resume_session_id.as_deref(),
None,
None,
&mut cancel_rx,
move |token| {
let mut buf = buffer_for_callback.lock().unwrap();
buf.push_str(token);
let paragraphs = drain_complete_paragraphs(&mut buf);
for chunk in paragraphs {
sent_any_chunk_for_callback.store(true, Ordering::Relaxed);
let _ = msg_tx_for_callback.send(chunk);
}
run_llm_turn(
LlmTurnContext {
transport: Arc::clone(&ctx.transport),
key: sender,
user: sender,
services: &ctx.services,
history: &ctx.history,
history_size: ctx.history_size,
save_history: save_whatsapp_history,
},
|_thinking| {},
|_activity| {},
);
tokio::pin!(chat_fut);
// Register as a permission responder for the duration of this chat turn.
let (_perm_guard, mut perm_rx) = ctx.services.permission_registry.register();
let result = loop {
tokio::select! {
r = &mut chat_fut => break r,
Some(perm_fwd) = perm_rx.recv() => {
let prompt_msg = format!(
"*Permission Request*\n\nTool: `{}`\n```json\n{}\n```\n\nReply *yes* to approve or *no* to deny.",
perm_fwd.tool_name,
serde_json::to_string_pretty(&perm_fwd.tool_input)
.unwrap_or_else(|_| perm_fwd.tool_input.to_string()),
);
let formatted = markdown_to_whatsapp(&prompt_msg);
for part in chunk_for_whatsapp(&formatted) {
let _ = ctx.transport.send_message(sender, &part, "").await;
}
// Store the response sender so the incoming message handler
// can resolve it when the user replies yes/no. Keyed by
// request_id (not just sender) so a second concurrent
// request doesn't drop the first's sender.
ctx.services.pending_perm_replies
.insert(sender.to_string(), perm_fwd.request_id.clone(), perm_fwd.response_tx)
.await;
// Spawn a timeout task: auto-deny if the user does not respond.
let pending = Arc::clone(&ctx.services.pending_perm_replies);
let timeout_sender = sender.to_string();
let timeout_request_id = perm_fwd.request_id.clone();
let timeout_transport = Arc::clone(&ctx.transport);
let timeout_secs = ctx.services.permission_timeout_secs;
tokio::spawn(async move {
tokio::time::sleep(std::time::Duration::from_secs(timeout_secs)).await;
if let Some(tx) = pending.remove_by_request_id(&timeout_sender, &timeout_request_id).await {
let _ = tx.send(PermissionDecision::Deny);
let msg = "Permission request timed out — denied (fail-closed).";
let _ = timeout_transport.send_message(&timeout_sender, msg, "").await;
}
});
}
}
};
// Flush remaining text.
let remaining = buffer.lock().unwrap().trim().to_string();
let did_send_any = sent_any_chunk.load(Ordering::Relaxed);
let (assistant_reply, new_session_id) = match result {
Ok(ClaudeCodeResult {
messages,
session_id,
..
}) => {
let reply = if !remaining.is_empty() {
let _ = msg_tx.send(remaining.clone());
remaining
} else if !did_send_any {
let last_text = messages
.iter()
.rev()
.find(|m| m.role == crate::llm::types::Role::Assistant && !m.content.is_empty())
.map(|m| m.content.clone())
.unwrap_or_default();
if !last_text.is_empty() {
let _ = msg_tx.send(last_text.clone());
}
last_text
} else {
remaining
};
slog!("[whatsapp] session_id from chat_stream: {:?}", session_id);
(reply, session_id)
}
Err(e) => {
slog!("[whatsapp] LLM error: {e}");
let err_msg = if let Some(url) = crate::llm::oauth::extract_login_url_from_error(&e) {
format!("Authentication required. Log in to Claude here: {url}")
} else {
format!("Error processing your request: {e}")
};
let _ = msg_tx.send(err_msg.clone());
(err_msg, None)
}
};
// Signal the posting task to finish and wait for it.
drop(msg_tx);
let _ = post_task.await;
// Record this exchange in conversation history.
if !assistant_reply.starts_with("Error processing") {
let mut guard = ctx.history.lock().await;
let conv = guard.entry(sender.to_string()).or_default();
if new_session_id.is_some() {
conv.session_id = new_session_id;
}
conv.entries.push(ConversationEntry {
role: ConversationRole::User,
sender: sender.to_string(),
content: user_message.to_string(),
});
conv.entries.push(ConversationEntry {
role: ConversationRole::Assistant,
sender: String::new(),
content: assistant_reply,
});
// Trim to configured maximum.
if conv.entries.len() > ctx.history_size {
let excess = conv.entries.len() - ctx.history_size;
conv.entries.drain(..excess);
}
save_whatsapp_history(&ctx.services.project_root, &guard);
}
WhatsAppFormat,
user_message,
)
.await;
}
// ── Tests ───────────────────────────────────────────────────────────────
+15
View File
@@ -155,6 +155,12 @@ pub struct ProjectConfig {
/// (disabled) — the on-demand `gc` MCP tool remains available regardless.
#[serde(default)]
pub gc_min_free_gb: u64,
/// Number of CRDT ops applied (locally or received from sync peers)
/// between periodic snapshot checkpoints (story 1249). Keeps the
/// replayed tail on startup bounded instead of growing forever between
/// the one-time post-replay snapshot and the next restart. Default: 5000.
#[serde(default = "default_snapshot_interval_ops")]
pub snapshot_interval_ops: usize,
}
/// Configuration for the filesystem watcher's sweep behaviour.
@@ -255,6 +261,11 @@ fn default_max_mesh_peers() -> usize {
3
}
/// Default number of ops between periodic CRDT snapshot checkpoints (story 1249).
pub fn default_snapshot_interval_ops() -> usize {
5000
}
/// Configuration for the low-disk-space watchdog's free-space thresholds.
///
/// Sleds check free space on the `/workspace` filesystem each tick and
@@ -472,6 +483,7 @@ impl Default for ProjectConfig {
status_push_enabled: default_status_push_enabled(),
merge_failure_block_threshold: default_merge_failure_block_threshold(),
gc_min_free_gb: 0,
snapshot_interval_ops: default_snapshot_interval_ops(),
}
}
}
@@ -564,6 +576,7 @@ impl ProjectConfig {
status_push_enabled: default_status_push_enabled(),
merge_failure_block_threshold: default_merge_failure_block_threshold(),
gc_min_free_gb: 0,
snapshot_interval_ops: default_snapshot_interval_ops(),
};
validate_agents(&config.agent)?;
return Ok(config);
@@ -607,6 +620,7 @@ impl ProjectConfig {
status_push_enabled: default_status_push_enabled(),
merge_failure_block_threshold: default_merge_failure_block_threshold(),
gc_min_free_gb: 0,
snapshot_interval_ops: default_snapshot_interval_ops(),
};
validate_agents(&config.agent)?;
Ok(config)
@@ -638,6 +652,7 @@ impl ProjectConfig {
status_push_enabled: default_status_push_enabled(),
merge_failure_block_threshold: default_merge_failure_block_threshold(),
gc_min_free_gb: 0,
snapshot_interval_ops: default_snapshot_interval_ops(),
})
}
}
+15
View File
@@ -11,6 +11,21 @@ fn default_config_when_missing() {
assert!(config.component.is_empty());
}
#[test]
fn snapshot_interval_ops_defaults_to_5000() {
let config = ProjectConfig::default();
assert_eq!(config.snapshot_interval_ops, 5000);
}
#[test]
fn snapshot_interval_ops_overridable() {
let toml_str = r#"
snapshot_interval_ops = 250
"#;
let config = ProjectConfig::parse(toml_str).unwrap();
assert_eq!(config.snapshot_interval_ops, 250);
}
#[test]
fn parse_multi_agent_toml() {
let toml_str = r#"
+1 -1
View File
@@ -49,7 +49,7 @@ pub use read::{
read_all_items, read_item, tombstoned_ids,
};
pub(crate) use state::flush_persistence;
pub use state::{init, subscribe};
pub use state::{checkpoint_on_shutdown, init, subscribe};
pub use types::{
ActiveAgentCrdt, ActiveAgentView, AgentThrottleCrdt, AgentThrottleView, CrdtEvent, EpicId,
EventLogEntryCrdt, GatewayConfigCrdt, GatewayProjectCrdt, GatewayProjectView, LlmSessionCrdt,
+165 -19
View File
@@ -24,7 +24,7 @@ use super::indices::{
rebuild_index, rebuild_llm_session_index, rebuild_merge_job_index, rebuild_node_index,
rebuild_test_job_index, rebuild_token_index,
};
use super::statics::{ALL_OPS, CRDT_EVENT_TX, PERSIST_PENDING, SYNC_TX, VECTOR_CLOCK};
use super::statics::{self, ALL_OPS, CRDT_EVENT_TX, PERSIST_PENDING, SYNC_TX, VECTOR_CLOCK};
use super::{CRDT_STATE, CrdtState};
use crate::slog;
@@ -41,8 +41,11 @@ pub(crate) enum PersistMsg {
/// Opens the SQLite database, loads or creates a node keypair, replays any
/// persisted ops to reconstruct state, and spawns a background persistence
/// task. Safe to call only once; subsequent calls are no-ops.
///
/// `snapshot_interval_ops` (story 1249) is the number of ops applied between
/// periodic snapshot checkpoints — see `ProjectConfig::snapshot_interval_ops`.
#[allow(clippy::string_slice)] // op_id is hex::encode output (ASCII-only), &op_id[..12] is always valid
pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> {
pub async fn init(db_path: &Path, snapshot_interval_ops: usize) -> Result<(), sqlx::Error> {
if CRDT_STATE.get().is_some() {
return Ok(());
}
@@ -52,6 +55,8 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> {
.create_if_missing(true);
let pool = SqlitePool::connect_with(options).await?;
sqlx::migrate!("./migrations").run(&pool).await?;
let _ = statics::CRDT_POOL.set(pool.clone());
let _ = statics::SNAPSHOT_INTERVAL_OPS.set(snapshot_interval_ops);
// Load or create the node keypair.
let keypair = load_or_create_keypair(&pool).await?;
@@ -73,15 +78,19 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> {
);
let kp = keypair.clone();
let restore_result = tokio::task::spawn_blocking(move || {
let doc: PipelineDoc = serde_json::from_str(&state_json)
.map_err(|e| format!("snapshot deserialize failed: {e}"))?;
let doc: PipelineDoc = match serde_json::from_str(&state_json) {
Ok(doc) => doc,
Err(e) => {
eprintln!("\n{}\n", snapshot_load_abort_message(&e.to_string()));
std::process::exit(1);
}
};
let mut crdt = BaseCrdt::<PipelineDoc>::new(&kp);
crdt.doc = doc;
Ok::<_, String>(crdt)
crdt
})
.await
.map_err(|e| sqlx::Error::Protocol(format!("snapshot restore panicked: {e}")))?
.map_err(sqlx::Error::Protocol)?;
.map_err(|e| sqlx::Error::Protocol(format!("snapshot restore panicked: {e}")))?;
// Replay only ops that arrived after the snapshot.
let tail_rows: Vec<(String,)> =
@@ -270,6 +279,8 @@ pub async fn init(db_path: &Path) -> Result<(), sqlx::Error> {
slog!("[crdt] Failed to persist op {}: {e}", &op_id[..12]);
}
PERSIST_PENDING.fetch_sub(1, Ordering::Relaxed);
maybe_periodic_checkpoint().await;
}
PersistMsg::Flush(reply) => {
// All ops queued before this message have already been processed.
@@ -375,21 +386,27 @@ async fn load_or_create_keypair(pool: &SqlitePool) -> Result<Ed25519KeyPair, sql
/// and back up the database file. Errors are logged but not propagated —
/// a failed snapshot just means the next restart will do a full replay.
async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, lamport_floor: u64) {
// Find the highest rowid currently in crdt_ops — ops with rowid <= this
// value are already captured in the snapshot.
let max_rowid: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(rowid), 0) FROM crdt_ops")
.fetch_one(pool)
.await
.unwrap_or(0);
let doc_ref = &crdt.doc;
let json = match serde_json::to_string(doc_ref) {
let json = match serde_json::to_string(&crdt.doc) {
Ok(j) => j,
Err(e) => {
slog!("[crdt] Failed to serialize snapshot: {e}");
return;
}
};
save_snapshot_json(pool, &json, lamport_floor).await;
}
/// Shared snapshot-writing body: back up the DB file and write `json` (an
/// already-serialized `PipelineDoc`) into the `crdt_snapshot` table. Used by
/// both the one-time post-replay snapshot ([`save_snapshot`]) and periodic /
/// shutdown checkpoints ([`checkpoint_now`]) added by story 1249.
async fn save_snapshot_json(pool: &SqlitePool, json: &str, at_seq: u64) {
// Find the highest rowid currently in crdt_ops — ops with rowid <= this
// value are already captured in the snapshot.
let max_rowid: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(rowid), 0) FROM crdt_ops")
.fetch_one(pool)
.await
.unwrap_or(0);
let json_len = json.len();
let now = chrono::Utc::now().to_rfc3339();
@@ -408,9 +425,9 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, lamport_
"INSERT OR REPLACE INTO crdt_snapshot (id, at_seq, max_rowid, state_json, created_at) \
VALUES (1, ?1, ?2, ?3, ?4)",
)
.bind(lamport_floor as i64)
.bind(at_seq as i64)
.bind(max_rowid)
.bind(&json)
.bind(json)
.bind(&now)
.execute(pool)
.await;
@@ -418,7 +435,7 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, lamport_
match result {
Ok(_) => slog!(
"[crdt] Snapshot saved: at_seq={}, max_rowid={}, json={}B",
lamport_floor,
at_seq,
max_rowid,
json_len
),
@@ -426,6 +443,86 @@ async fn save_snapshot(pool: &SqlitePool, crdt: &BaseCrdt<PipelineDoc>, lamport_
}
}
/// Pure threshold decision for periodic checkpoints (story 1249 AC2/AC3).
///
/// Returns `true` only when at least one op has been applied since the last
/// checkpoint AND that count has reached `threshold`. The `ops_since_last >
/// 0` guard is what makes AC3 (skip when nothing changed) hold even for a
/// `threshold` of 0 or 1.
fn should_checkpoint(ops_since_last: usize, threshold: usize) -> bool {
ops_since_last > 0 && ops_since_last >= threshold
}
/// Called after every op the persistence task writes. Triggers a checkpoint
/// once `OPS_SINCE_SNAPSHOT` reaches the configured `SNAPSHOT_INTERVAL_OPS`
/// (story 1249 AC1/AC2).
async fn maybe_periodic_checkpoint() {
let threshold = statics::SNAPSHOT_INTERVAL_OPS
.get()
.copied()
.unwrap_or(crate::config::default_snapshot_interval_ops());
let ops_since_last = statics::OPS_SINCE_SNAPSHOT.load(Ordering::Relaxed);
if should_checkpoint(ops_since_last, threshold) {
checkpoint_now("periodic").await;
}
}
/// Take a snapshot of the current CRDT state, unless no ops have been
/// applied since the last checkpoint (story 1249 AC3). Resets
/// `OPS_SINCE_SNAPSHOT` back to zero afterwards. `reason` is only used for
/// the log line (e.g. `"periodic"` or `"shutdown"`).
async fn checkpoint_now(reason: &str) {
if statics::OPS_SINCE_SNAPSHOT.load(Ordering::Relaxed) == 0 {
return;
}
let Some(pool) = statics::CRDT_POOL.get() else {
return;
};
let Some(state_mutex) = super::get_crdt() else {
return;
};
let json = {
let Ok(state) = state_mutex.lock() else {
return;
};
match serde_json::to_string(&state.crdt.doc) {
Ok(j) => j,
Err(e) => {
slog!("[crdt] Failed to serialize {reason} checkpoint: {e}");
return;
}
}
};
let at_seq = statics::LAST_SEQ.load(Ordering::Relaxed);
save_snapshot_json(pool, &json, at_seq).await;
statics::OPS_SINCE_SNAPSHOT.store(0, Ordering::Relaxed);
slog!("[crdt] Checkpoint ({reason}) complete");
}
/// Take a final checkpoint on clean shutdown (story 1249 AC1), if any ops
/// have been applied since the last one. Safe to call even if the CRDT
/// layer was never initialised — it's a no-op in that case.
pub async fn checkpoint_on_shutdown() {
checkpoint_now("shutdown").await;
}
/// Build the abort message printed when a CRDT snapshot fails to deserialize
/// into the current schema. `deserialize_error` is the `Display` text of the
/// `serde_json::Error`, which names the missing/mismatched field; it is
/// included verbatim so the operator can see exactly what changed.
fn snapshot_load_abort_message(deserialize_error: &str) -> String {
format!(
"error: failed to load the CRDT snapshot (crdt_snapshot row):\n \
{deserialize_error}\n\n\
This snapshot was written by a binary with a different schema and is \
missing a field the current binary expects.\n\
No data has been lost: the ops log (crdt_ops) is intact and is the \
durable source of truth the snapshot is only a fast-path replay cache.\n\
To recover: delete the crdt_snapshot row (e.g. `DELETE FROM crdt_snapshot \
WHERE id = 1;`) so init.rs rebuilds state from crdt_ops on the next start."
)
}
/// Extract the filesystem path from a SqlitePool's connect options.
fn pool_path(pool: &SqlitePool) -> Option<String> {
use sqlx::ConnectOptions;
@@ -433,3 +530,52 @@ fn pool_path(pool: &SqlitePool) -> Option<String> {
let filename = opts.get_filename();
filename.to_str().map(|s| s.to_string())
}
#[cfg(test)]
mod tests {
use super::{should_checkpoint, snapshot_load_abort_message};
/// AC3: a checkpoint must never fire when nothing has changed, no matter
/// how low the configured threshold is.
#[test]
fn should_checkpoint_false_when_no_ops_applied() {
assert!(!should_checkpoint(0, 5000));
assert!(!should_checkpoint(0, 1));
assert!(!should_checkpoint(0, 0));
}
/// AC2: the threshold is whatever N is configured to, not a hardcoded value.
#[test]
fn should_checkpoint_respects_configured_threshold() {
assert!(!should_checkpoint(4999, 5000));
assert!(should_checkpoint(5000, 5000));
assert!(should_checkpoint(5001, 5000));
assert!(should_checkpoint(1, 1));
}
/// The abort message must name the failing field (via the serde error
/// text), explain that the ops log is intact, and give the exact
/// recovery step — this is what the operator reads at 3am.
#[test]
fn snapshot_load_abort_message_names_field_and_explains_recovery() {
let err = "missing field `gateway_config` at line 1 column 4821";
let msg = snapshot_load_abort_message(err);
assert!(
msg.contains("missing field `gateway_config`"),
"message must name the failing field verbatim: {msg}"
);
assert!(
msg.contains("different schema"),
"message must explain the schema-mismatch cause: {msg}"
);
assert!(
msg.contains("crdt_ops") && msg.contains("intact"),
"message must state the ops log is intact and no data is lost: {msg}"
);
assert!(
msg.contains("DELETE FROM crdt_snapshot"),
"message must give the exact recovery step: {msg}"
);
}
}
+1 -1
View File
@@ -26,8 +26,8 @@ mod tests;
// ── Re-exports for crdt_state siblings ──────────────────────────────
pub use init::init;
pub(crate) use init::{PersistMsg, flush_persistence};
pub use init::{checkpoint_on_shutdown, init};
/// Subscribe to CRDT state-transition events.
///
+29 -1
View File
@@ -10,10 +10,11 @@
//! tests do not share `ALL_OPS` — preventing one test's `apply_compaction`
//! from pruning another test's freshly-written ops.
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::{AtomicU64, AtomicUsize};
use std::sync::{Mutex, OnceLock};
use bft_json_crdt::json_crdt::SignedOp;
use sqlx::SqlitePool;
use tokio::sync::broadcast;
use super::super::VectorClock;
@@ -46,6 +47,31 @@ pub(crate) static ALL_OPS: OnceLock<Mutex<Vec<String>>> = OnceLock::new();
/// re-parsing all ops when a peer requests `our_vector_clock()`.
pub(crate) static VECTOR_CLOCK: OnceLock<Mutex<VectorClock>> = OnceLock::new();
/// The CRDT SQLite pool, stashed here so periodic and shutdown checkpoints
/// (story 1249) can reuse it without threading it through every call site
/// that might trigger one. Set once, at the end of `init::init()`.
pub(crate) static CRDT_POOL: OnceLock<SqlitePool> = OnceLock::new();
/// Number of ops between periodic snapshot checkpoints (story 1249).
/// Set once, from `ProjectConfig::snapshot_interval_ops`, at the end of
/// `init::init()`. Falls back to `config::default_snapshot_interval_ops()`
/// (5000) if `init()` hasn't set it yet.
pub(crate) static SNAPSHOT_INTERVAL_OPS: OnceLock<usize> = OnceLock::new();
/// Count of ops applied (locally created or received from sync peers) since
/// the last snapshot checkpoint. Incremented in [`track_op`]; reset to zero
/// after a checkpoint is taken. A checkpoint is skipped when this is zero
/// (story 1249 AC3) so restarting the server or an idle period never writes
/// a redundant snapshot.
pub(crate) static OPS_SINCE_SNAPSHOT: AtomicUsize = AtomicUsize::new(0);
/// Highest op sequence number seen since startup, updated in [`track_op`].
/// Used as the informational `at_seq` value for periodic/shutdown
/// checkpoints (the replay-boundary correctness depends only on
/// `max_rowid`, computed fresh from SQLite at snapshot time — this is just
/// for accurate logging/debugging).
pub(crate) static LAST_SEQ: AtomicU64 = AtomicU64::new(0);
#[cfg(test)]
thread_local! {
/// Per-thread op journal for test isolation. Each test thread sees its
@@ -107,4 +133,6 @@ pub(in crate::crdt_state) fn track_op(signed: &SignedOp, json: String) {
let author_hex = hex::encode(&signed.author());
*clock.entry(author_hex).or_insert(0) += 1;
}
OPS_SINCE_SNAPSHOT.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
LAST_SEQ.fetch_max(signed.inner.seq, std::sync::atomic::Ordering::Relaxed);
}
+10 -1
View File
@@ -94,7 +94,9 @@ pub async fn run(config_path: &Path, port: u16) -> Result<(), std::io::Error> {
// Initialise the CRDT so gateway_config.active_project is persisted across restarts.
let crdt_db = config_dir.join("gateway.db");
if let Err(e) = crate::crdt_state::init(&crdt_db).await {
if let Err(e) =
crate::crdt_state::init(&crdt_db, crate::config::default_snapshot_interval_ops()).await
{
crate::slog!(
"[gateway] Warning: CRDT init failed ({e}); active-project selection will not persist"
);
@@ -146,6 +148,13 @@ pub async fn run(config_path: &Path, port: u16) -> Result<(), std::io::Error> {
.run(route)
.await;
// Story 1249: take a final CRDT snapshot on clean shutdown (mirrors main.rs).
crate::crdt_state::checkpoint_on_shutdown().await;
// Story 1249: take a final CRDT snapshot on clean shutdown so the next
// restart's replayed tail stays bounded.
crate::crdt_state::checkpoint_on_shutdown().await;
// Best-effort shutdown notification: signal the Matrix bot so it can post
// "going offline" before the process exits. Mirror of main.rs:346.
{
-1
View File
@@ -1211,7 +1211,6 @@ async fn handle_project_rebuild_tool(
drain_timeout_secs,
force,
&state.projects,
Some(state.port),
&state.config_dir,
)
.await;
@@ -1,7 +1,7 @@
//! Story field update and unblock tools.
use crate::http::context::AppContext;
use crate::http::workflow::update_story_in_file;
use crate::http::workflow::{is_bug_item, update_bug_fields_in_file, update_story_in_file};
use crate::slog_warn;
use crate::validation::UpdateStoryRequest;
use serde_json::Value;
@@ -14,6 +14,26 @@ pub(crate) fn tool_update_story(args: &Value, ctx: &AppContext) -> Result<String
let req = UpdateStoryRequest::from_json(args)?;
// Story 1250: steps_to_reproduce/actual_result/expected_result only exist on
// bug items. Reject up front rather than silently creating the section on a
// story/spike/refactor that has no such field.
if (req.steps_to_reproduce.is_some()
|| req.actual_result.is_some()
|| req.expected_result.is_some())
&& !is_bug_item(story_id)
{
let field = if req.steps_to_reproduce.is_some() {
"steps_to_reproduce"
} else if req.actual_result.is_some() {
"actual_result"
} else {
"expected_result"
};
return Err(format!(
"Field '{field}' is only valid on bug items. '{story_id}' is not a bug."
));
}
// Explicit top-level args map onto typed CRDT registers directly (story 929:
// no YAML front-matter writes). The `front_matter` object is the legacy
// escape hatch; every known key is recognised and routed below, and any
@@ -194,6 +214,19 @@ pub(crate) fn tool_update_story(args: &Value, ctx: &AppContext) -> Result<String
)?;
}
if req.steps_to_reproduce.is_some()
|| req.actual_result.is_some()
|| req.expected_result.is_some()
{
update_bug_fields_in_file(
&root,
story_id,
req.steps_to_reproduce.as_ref().map(|d| d.as_str()),
req.actual_result.as_ref().map(|d| d.as_str()),
req.expected_result.as_ref().map(|d| d.as_str()),
)?;
}
// Bug 503: warn if any depends_on in the (now updated) story points at an archived story.
// Story 929: reads from the CRDT (was a FS-yaml scan).
let archived_deps = crate::crdt_state::check_archived_deps_crdt(story_id);
@@ -264,6 +264,18 @@ pub(super) fn story_tools() -> Vec<Value> {
"type": "string",
"description": "New description text to replace the '## Description' section content"
},
"steps_to_reproduce": {
"type": "string",
"description": "Bug items only: replace the '## How to Reproduce' section content. Errors if story_id is not a bug."
},
"actual_result": {
"type": "string",
"description": "Bug items only: replace the '## Actual Result' section content. Errors if story_id is not a bug."
},
"expected_result": {
"type": "string",
"description": "Bug items only: replace the '## Expected Result' section content. Errors if story_id is not a bug."
},
"agent": {
"type": "string",
"description": "Set or change the 'agent' YAML front matter field"
+63 -2
View File
@@ -2,7 +2,10 @@
use std::path::Path;
use super::super::create_item_in_backlog;
use super::super::{
create_item_in_backlog, create_section_content, read_story_content, replace_section_content,
story_stage, write_story_content,
};
/// Create a bug file and store it in the database.
///
@@ -66,11 +69,69 @@ pub fn create_bug_file(
)
}
/// Update the `## How to Reproduce`, `## Actual Result` and/or `## Expected
/// Result` sections of an existing bug (story 1250).
///
/// At least one of the three must be provided. Callers must confirm the
/// target item is a bug (via [`is_bug_item`]) before calling this — it does
/// not itself check item type, so calling it against a non-bug item will
/// silently create these sections.
pub fn update_bug_fields_in_file(
project_root: &Path,
bug_id: &str,
steps_to_reproduce: Option<&str>,
actual_result: Option<&str>,
expected_result: Option<&str>,
) -> Result<(), String> {
if steps_to_reproduce.is_none() && actual_result.is_none() && expected_result.is_none() {
return Err(
"At least one of 'steps_to_reproduce', 'actual_result' or 'expected_result' \
must be provided."
.to_string(),
);
}
let mut contents = read_story_content(project_root, bug_id)?;
if let Some(steps) = steps_to_reproduce {
contents = match replace_section_content(&contents, "How to Reproduce", steps) {
Ok(updated) => updated,
Err(_) => {
create_section_content(&contents, "How to Reproduce", steps, Some("Actual Result"))
}
};
}
if let Some(actual) = actual_result {
contents = match replace_section_content(&contents, "Actual Result", actual) {
Ok(updated) => updated,
Err(_) => {
create_section_content(&contents, "Actual Result", actual, Some("Expected Result"))
}
};
}
if let Some(expected) = expected_result {
contents = match replace_section_content(&contents, "Expected Result", expected) {
Ok(updated) => updated,
Err(_) => create_section_content(
&contents,
"Expected Result",
expected,
Some("Acceptance Criteria"),
),
};
}
let stage = story_stage(bug_id).unwrap_or_else(|| "1_backlog".to_string());
write_story_content(project_root, bug_id, &stage, &contents, None);
Ok(())
}
/// Returns true if the item stem is a bug item.
///
/// Checks the slug-based ID format first (e.g. `"4_bug_login_crash"`), then
/// consults the typed CRDT `item_type` register for numeric-only IDs (story 933).
pub(super) fn is_bug_item(stem: &str) -> bool {
pub fn is_bug_item(stem: &str) -> bool {
let after_num = stem.trim_start_matches(|c: char| c.is_ascii_digit());
if after_num.starts_with("_bug_") {
return true;
+1 -1
View File
@@ -8,7 +8,7 @@ mod spike;
#[cfg(test)]
mod tests;
pub use bug::{create_bug_file, list_bug_files};
pub use bug::{create_bug_file, is_bug_item, list_bug_files, update_bug_fields_in_file};
pub use epic::create_epic_file;
pub use refactor::{create_refactor_file, list_refactor_files};
pub use spike::create_spike_file;
+2 -2
View File
@@ -6,8 +6,8 @@ mod test_results;
mod utils;
pub use bug_ops::{
create_bug_file, create_epic_file, create_refactor_file, create_spike_file, list_bug_files,
list_refactor_files,
create_bug_file, create_epic_file, create_refactor_file, create_spike_file, is_bug_item,
list_bug_files, list_refactor_files, update_bug_fields_in_file,
};
pub use pipeline::{
PipelineState, UpcomingStory, load_pipeline_state, load_upcoming_stories, validate_story_dirs,
+5
View File
@@ -491,6 +491,11 @@ async fn main() -> Result<(), std::io::Error> {
let result = Server::new(TcpListener::bind(&addr)).run(app).await;
// Story 1249: take a final CRDT snapshot on clean shutdown so the next
// restart's replayed tail stays bounded, even if fewer than
// `snapshot_interval_ops` ops have accumulated since the last periodic one.
crate::crdt_state::checkpoint_on_shutdown().await;
// ── Shutdown notifications (best-effort) ──────────────────────────────────
startup::bots::notify_shutdown(&bot_ctxs).await;
+4 -1
View File
@@ -16,7 +16,10 @@ pub mod summary;
#[allow(unused_imports)]
pub use status::format_merge_status_message;
#[allow(unused_imports)]
pub use summary::{summarize_merge_failure_kind, summarize_merge_result};
pub use summary::{
bound_plain_error, bound_report_for_storage, summarize_merge_failure_kind,
summarize_merge_result,
};
// ── Error type ────────────────────────────────────────────────────────────────
+186
View File
@@ -113,6 +113,78 @@ pub fn summarize_merge_failure_kind(kind: &MergeFailureKind) -> String {
summarize(failing_gate_label_for_kind(kind), &kind.to_gate_output())
}
/// Return a copy of `result` with its embedded output string (`gate_output`
/// on [`MergeResult::Success`], `output` on every other variant) bounded to
/// [`SUMMARY_TAIL_LINES`].
///
/// Used before storing a [`MergeResult`] in the replicated CRDT `merge_jobs`
/// collection: the full untruncated text already lives on disk (see
/// `service::merge::io::write_merge_report` and `MergeReport::report_path`),
/// so only a bounded tail needs to travel through CRDT sync.
pub fn bound_result_for_storage(result: &MergeResult) -> MergeResult {
match result {
MergeResult::Success {
conflicts_resolved,
conflict_details,
gate_output,
} => MergeResult::Success {
conflicts_resolved: *conflicts_resolved,
conflict_details: conflict_details.clone(),
gate_output: truncate_gate_output(gate_output, SUMMARY_TAIL_LINES),
},
MergeResult::Conflict { details, output } => MergeResult::Conflict {
details: details.clone(),
output: truncate_gate_output(output, SUMMARY_TAIL_LINES),
},
MergeResult::GateFailure {
output,
failure_kind,
} => MergeResult::GateFailure {
output: truncate_gate_output(output, SUMMARY_TAIL_LINES),
failure_kind: failure_kind.clone(),
},
MergeResult::NoCommits { output } => MergeResult::NoCommits {
output: truncate_gate_output(output, SUMMARY_TAIL_LINES),
},
MergeResult::Other {
output,
conflict_details,
} => MergeResult::Other {
output: truncate_gate_output(output, SUMMARY_TAIL_LINES),
conflict_details: conflict_details.clone(),
},
}
}
/// Return a copy of `report` whose embedded [`MergeResult`] output is bounded
/// via [`bound_result_for_storage`]; `report_path` (the pointer to the full
/// untruncated text on disk) and every other field are carried over as-is.
pub fn bound_report_for_storage(
report: &crate::agents::merge::MergeReport,
) -> crate::agents::merge::MergeReport {
crate::agents::merge::MergeReport {
story_id: report.story_id.clone(),
result: bound_result_for_storage(&report.result),
worktree_cleaned_up: report.worktree_cleaned_up,
story_archived: report.story_archived,
report_path: report.report_path.clone(),
}
}
/// Bound a plain (non-[`MergeResult`]) error string for CRDT storage,
/// appending a pointer to the full text on disk when `report_path` is given.
///
/// Used for the hard-error path (git/process failures that short-circuit
/// before a [`MergeResult`] is ever constructed), so the CRDT `merge_jobs`
/// error field stays bounded the same way completed-job entries do.
pub fn bound_plain_error(error: &str, report_path: Option<&str>) -> String {
let truncated = truncate_gate_output(error, SUMMARY_TAIL_LINES);
match report_path {
Some(path) => format!("{truncated}\n\nFull output: {path}"),
None => truncated,
}
}
#[cfg(test)]
mod tests {
use super::*;
@@ -236,4 +308,118 @@ mod tests {
let summary = summarize_merge_failure_kind(&kind);
assert!(!summary.contains("Quality gates failed:"));
}
// ── bound_result_for_storage ────────────────────────────────────────────────
fn long_output(n: usize) -> String {
(1..=n)
.map(|i| format!("line{i}"))
.collect::<Vec<_>>()
.join("\n")
}
#[test]
fn bound_result_for_storage_truncates_gate_failure_output() {
let result = MergeResult::GateFailure {
output: long_output(100),
failure_kind: Some(GateFailureKind::Test),
};
let bounded = bound_result_for_storage(&result);
assert!(bounded.output().len() < result.output().len());
assert!(bounded.output().contains("line100"));
assert!(matches!(
bounded,
MergeResult::GateFailure {
failure_kind: Some(GateFailureKind::Test),
..
}
));
}
#[test]
fn bound_result_for_storage_truncates_success_gate_output() {
let result = MergeResult::Success {
conflicts_resolved: true,
conflict_details: Some("resolved automatically".to_string()),
gate_output: long_output(100),
};
let bounded = bound_result_for_storage(&result);
assert!(bounded.output().len() < result.output().len());
assert!(bounded.output().contains("line100"));
match bounded {
MergeResult::Success {
conflicts_resolved,
conflict_details,
..
} => {
assert!(conflicts_resolved);
assert_eq!(conflict_details.as_deref(), Some("resolved automatically"));
}
other => panic!("expected Success, got {other:?}"),
}
}
#[test]
fn bound_result_for_storage_leaves_short_output_unchanged() {
let result = MergeResult::NoCommits {
output: "no commits to merge".to_string(),
};
let bounded = bound_result_for_storage(&result);
assert_eq!(bounded.output(), "no commits to merge");
}
// ── bound_report_for_storage ────────────────────────────────────────────────
#[test]
fn bound_report_for_storage_truncates_output_and_keeps_report_path() {
let report = crate::agents::merge::MergeReport {
story_id: "42_story".to_string(),
result: MergeResult::GateFailure {
output: long_output(100),
failure_kind: Some(GateFailureKind::Test),
},
worktree_cleaned_up: false,
story_archived: false,
report_path: Some(".huskies/merge_reports/42_story-123.log".to_string()),
};
let bounded = bound_report_for_storage(&report);
assert!(bounded.result.output().len() < report.result.output().len());
assert_eq!(
bounded.report_path.as_deref(),
report.report_path.as_deref()
);
assert_eq!(bounded.story_id, report.story_id);
// Round-trips through the same JSON shape `get_merge_status` expects.
let json = serde_json::to_string(&bounded).unwrap();
let decoded: crate::agents::merge::MergeReport = serde_json::from_str(&json).unwrap();
assert!(decoded.result.output().contains("line100"));
assert_eq!(
decoded.report_path.as_deref(),
Some(".huskies/merge_reports/42_story-123.log")
);
}
// ── bound_plain_error ────────────────────────────────────────────────────────
#[test]
fn bound_plain_error_appends_pointer_when_path_given() {
let bounded = bound_plain_error("boom", Some(".huskies/merge_reports/1_story-1.log"));
assert!(bounded.contains("boom"));
assert!(bounded.contains("Full output: .huskies/merge_reports/1_story-1.log"));
}
#[test]
fn bound_plain_error_no_pointer_when_no_path() {
let bounded = bound_plain_error("boom", None);
assert_eq!(bounded, "boom");
}
#[test]
fn bound_plain_error_truncates_long_error() {
let error = long_output(100);
let bounded = bound_plain_error(&error, None);
assert!(bounded.len() < error.len());
assert!(bounded.contains("line100"));
}
}
+9 -1
View File
@@ -338,7 +338,15 @@ pub(crate) async fn init_subsystems(app_state: &Arc<SessionState>, cwd: &Path, i
let huskies_dir = db_path.parent().unwrap_or(db_path);
migrate_json_stores_to_sqlite(huskies_dir).await;
}
if let Err(e) = crdt_state::init(db_path).await {
let snapshot_interval_ops = app_state
.project_root
.lock()
.unwrap()
.as_ref()
.and_then(|root| config::ProjectConfig::load(root).ok())
.map(|cfg| cfg.snapshot_interval_ops)
.unwrap_or_else(config::default_snapshot_interval_ops);
if let Err(e) = crdt_state::init(db_path, snapshot_interval_ops).await {
crate::slog!("[crdt] Failed to initialise CRDT state layer: {e}");
} else {
crdt_state::migrate_names_from_slugs();
File diff suppressed because it is too large Load Diff
+1 -70
View File
@@ -169,80 +169,11 @@ pub fn format_report(report: &CleanupReport, confirm: bool) -> String {
#[cfg(test)]
mod tests {
use super::*;
use crate::config::WatcherConfig;
use chrono::Utc;
use crate::worktree::test_support::{done_stage, empty_config, init_git_repo};
use std::fs;
use std::path::PathBuf;
use std::process::Command;
use tempfile::TempDir;
fn init_git_repo(dir: &std::path::Path) {
use crate::git_test_support::git_ok;
git_ok(
Command::new("git").args(["init"]).current_dir(dir).output(),
"git init",
);
git_ok(
Command::new("git")
.args(["config", "user.email", "test@test.com"])
.current_dir(dir)
.output(),
"git config user.email",
);
git_ok(
Command::new("git")
.args(["config", "user.name", "Test"])
.current_dir(dir)
.output(),
"git config user.name",
);
git_ok(
Command::new("git")
.args(["commit", "--allow-empty", "-m", "init"])
.current_dir(dir)
.output(),
"git commit",
);
}
fn empty_config() -> ProjectConfig {
ProjectConfig {
component: vec![],
agent: vec![],
watcher: WatcherConfig::default(),
disk_watch: Default::default(),
default_qa: "server".to_string(),
default_coder_model: None,
max_coders: None,
max_retries: 2,
base_branch: None,
rate_limit_notifications: true,
web_ui_status_consumer: true,
matrix_status_consumer: true,
slack_status_consumer: true,
discord_status_consumer: true,
whatsapp_status_consumer: true,
timezone: None,
rendezvous: None,
trusted_keys: Vec::new(),
crdt_require_token: false,
crdt_tokens: Vec::new(),
max_mesh_peers: 3,
gateway_url: None,
gateway_project: None,
status_push_enabled: true,
merge_failure_block_threshold: 3,
gc_min_free_gb: 0,
}
}
fn done_stage() -> Stage {
Stage::Done {
merged_at: Utc::now(),
merge_commit: crate::pipeline_state::GitSha("abc123".to_string()),
}
}
async fn setup_project_with_real_worktree(story_id: &str) -> (TempDir, PathBuf) {
let tmp = TempDir::new().unwrap();
let project_root = tmp.path().join("project");
+2 -61
View File
@@ -210,71 +210,12 @@ pub(crate) async fn run_shell_command(cmd: &str, cwd: &Path) -> Result<(), Strin
#[cfg(test)]
mod tests {
use super::*;
use crate::config::{ComponentConfig, WatcherConfig};
use crate::config::ComponentConfig;
use crate::worktree::test_support::{empty_config, init_git_repo};
use std::fs;
use std::process::Command;
use tempfile::TempDir;
fn init_git_repo(dir: &Path) {
use crate::git_test_support::git_ok;
git_ok(
Command::new("git").args(["init"]).current_dir(dir).output(),
"git init",
);
git_ok(
Command::new("git")
.args(["config", "user.email", "test@test.com"])
.current_dir(dir)
.output(),
"git config user.email",
);
git_ok(
Command::new("git")
.args(["config", "user.name", "Test"])
.current_dir(dir)
.output(),
"git config user.name",
);
git_ok(
Command::new("git")
.args(["commit", "--allow-empty", "-m", "init"])
.current_dir(dir)
.output(),
"git commit",
);
}
fn empty_config() -> ProjectConfig {
ProjectConfig {
component: vec![],
agent: vec![],
watcher: WatcherConfig::default(),
disk_watch: Default::default(),
default_qa: "server".to_string(),
default_coder_model: None,
max_coders: None,
max_retries: 2,
base_branch: None,
rate_limit_notifications: true,
web_ui_status_consumer: true,
matrix_status_consumer: true,
slack_status_consumer: true,
discord_status_consumer: true,
whatsapp_status_consumer: true,
timezone: None,
rendezvous: None,
trusted_keys: Vec::new(),
crdt_require_token: false,
crdt_tokens: Vec::new(),
max_mesh_peers: 3,
gateway_url: None,
gateway_project: None,
status_push_enabled: true,
merge_failure_block_threshold: 3,
gc_min_free_gb: 0,
}
}
fn failing_setup_config() -> ProjectConfig {
ProjectConfig {
component: vec![ComponentConfig {
+1 -29
View File
@@ -232,38 +232,10 @@ pub fn migrate_slug_paths(project_root: &Path, migrations: &[(String, String)])
#[cfg(test)]
mod tests {
use super::*;
use crate::worktree::test_support::init_git_repo;
use std::fs;
use tempfile::TempDir;
fn init_git_repo(dir: &Path) {
use crate::git_test_support::git_ok;
git_ok(
Command::new("git").args(["init"]).current_dir(dir).output(),
"git init",
);
git_ok(
Command::new("git")
.args(["config", "user.email", "test@test.com"])
.current_dir(dir)
.output(),
"git config user.email",
);
git_ok(
Command::new("git")
.args(["config", "user.name", "Test"])
.current_dir(dir)
.output(),
"git config user.name",
);
git_ok(
Command::new("git")
.args(["commit", "--allow-empty", "-m", "init"])
.current_dir(dir)
.output(),
"git commit",
);
}
#[test]
fn branch_name_format() {
assert_eq!(branch_name("42_my_story"), "feature/story-42_my_story");
+4
View File
@@ -6,6 +6,10 @@ mod create;
mod git;
mod remove;
mod sweep;
/// Shared test fixtures (`init_git_repo`, `empty_config`) reused across the
/// worktree module's test suites and the pool worktree-lifecycle tests.
#[cfg(test)]
pub(crate) mod test_support;
pub use cleanup::{format_report, run_cleanup};
pub use create::create_worktree;
+1 -62
View File
@@ -68,71 +68,10 @@ pub async fn remove_worktree_by_story_id(
#[cfg(test)]
mod tests {
use super::*;
use crate::config::WatcherConfig;
use crate::worktree::test_support::{empty_config, init_git_repo};
use std::fs;
use std::process::Command;
use tempfile::TempDir;
fn init_git_repo(dir: &std::path::Path) {
use crate::git_test_support::git_ok;
git_ok(
Command::new("git").args(["init"]).current_dir(dir).output(),
"git init",
);
git_ok(
Command::new("git")
.args(["config", "user.email", "test@test.com"])
.current_dir(dir)
.output(),
"git config user.email",
);
git_ok(
Command::new("git")
.args(["config", "user.name", "Test"])
.current_dir(dir)
.output(),
"git config user.name",
);
git_ok(
Command::new("git")
.args(["commit", "--allow-empty", "-m", "init"])
.current_dir(dir)
.output(),
"git commit",
);
}
fn empty_config() -> ProjectConfig {
ProjectConfig {
component: vec![],
agent: vec![],
watcher: WatcherConfig::default(),
disk_watch: Default::default(),
default_qa: "server".to_string(),
default_coder_model: None,
max_coders: None,
max_retries: 2,
base_branch: None,
rate_limit_notifications: true,
web_ui_status_consumer: true,
matrix_status_consumer: true,
slack_status_consumer: true,
discord_status_consumer: true,
whatsapp_status_consumer: true,
timezone: None,
rendezvous: None,
trusted_keys: Vec::new(),
crdt_require_token: false,
crdt_tokens: Vec::new(),
max_mesh_peers: 3,
gateway_url: None,
gateway_project: None,
status_push_enabled: true,
merge_failure_block_threshold: 3,
gc_min_free_gb: 0,
}
}
#[tokio::test]
async fn remove_worktree_by_story_id_returns_err_when_not_found() {
let tmp = TempDir::new().unwrap();
+1 -69
View File
@@ -89,81 +89,13 @@ where
#[cfg(test)]
mod tests {
use super::*;
use crate::config::WatcherConfig;
use crate::worktree::test_support::{done_stage, empty_config, init_git_repo};
use chrono::Utc;
use std::fs;
use std::num::NonZeroU32;
use std::path::PathBuf;
use std::process::Command;
use tempfile::TempDir;
fn init_git_repo(dir: &Path) {
use crate::git_test_support::git_ok;
git_ok(
Command::new("git").args(["init"]).current_dir(dir).output(),
"git init",
);
git_ok(
Command::new("git")
.args(["config", "user.email", "test@test.com"])
.current_dir(dir)
.output(),
"git config user.email",
);
git_ok(
Command::new("git")
.args(["config", "user.name", "Test"])
.current_dir(dir)
.output(),
"git config user.name",
);
git_ok(
Command::new("git")
.args(["commit", "--allow-empty", "-m", "init"])
.current_dir(dir)
.output(),
"git commit",
);
}
fn empty_config() -> ProjectConfig {
ProjectConfig {
component: vec![],
agent: vec![],
watcher: WatcherConfig::default(),
disk_watch: Default::default(),
default_qa: "server".to_string(),
default_coder_model: None,
max_coders: None,
max_retries: 2,
base_branch: None,
rate_limit_notifications: true,
web_ui_status_consumer: true,
matrix_status_consumer: true,
slack_status_consumer: true,
discord_status_consumer: true,
whatsapp_status_consumer: true,
timezone: None,
rendezvous: None,
trusted_keys: Vec::new(),
crdt_require_token: false,
crdt_tokens: Vec::new(),
max_mesh_peers: 3,
gateway_url: None,
gateway_project: None,
status_push_enabled: true,
merge_failure_block_threshold: 3,
gc_min_free_gb: 0,
}
}
fn done_stage() -> Stage {
Stage::Done {
merged_at: Utc::now(),
merge_commit: crate::pipeline_state::GitSha("abc123".to_string()),
}
}
fn archived_stage() -> Stage {
Stage::Archived {
archived_at: Utc::now(),
+84
View File
@@ -0,0 +1,84 @@
//! Shared test fixtures for worktree module tests: a throwaway git repo and a
//! minimal [`ProjectConfig`] with no components, used by the worktree
//! create/remove/cleanup/sweep tests and the pool worktree-lifecycle
//! subscriber tests so each doesn't hand-roll its own copy.
use crate::config::{ProjectConfig, WatcherConfig};
use crate::git_test_support::git_ok;
use crate::pipeline_state::{GitSha, Stage};
use chrono::Utc;
use std::path::Path;
use std::process::Command;
/// Initialize a git repo at `dir` with test identity configured and a
/// committed empty initial commit, so worktree creation has a valid base
/// branch to fork from.
pub(crate) fn init_git_repo(dir: &Path) {
git_ok(
Command::new("git").args(["init"]).current_dir(dir).output(),
"git init",
);
git_ok(
Command::new("git")
.args(["config", "user.email", "test@test.com"])
.current_dir(dir)
.output(),
"git config user.email",
);
git_ok(
Command::new("git")
.args(["config", "user.name", "Test"])
.current_dir(dir)
.output(),
"git config user.name",
);
git_ok(
Command::new("git")
.args(["commit", "--allow-empty", "-m", "init"])
.current_dir(dir)
.output(),
"git commit",
);
}
/// A minimal [`ProjectConfig`] with no components, suitable for worktree
/// tests that don't exercise setup/teardown commands.
pub(crate) fn empty_config() -> ProjectConfig {
ProjectConfig {
component: vec![],
agent: vec![],
watcher: WatcherConfig::default(),
disk_watch: Default::default(),
default_qa: "server".to_string(),
default_coder_model: None,
max_coders: None,
max_retries: 2,
base_branch: None,
rate_limit_notifications: true,
web_ui_status_consumer: true,
matrix_status_consumer: true,
slack_status_consumer: true,
discord_status_consumer: true,
whatsapp_status_consumer: true,
timezone: None,
rendezvous: None,
trusted_keys: Vec::new(),
crdt_require_token: false,
crdt_tokens: Vec::new(),
max_mesh_peers: 3,
gateway_url: None,
gateway_project: None,
status_push_enabled: true,
merge_failure_block_threshold: 3,
gc_min_free_gb: 0,
snapshot_interval_ops: 5000,
}
}
/// A `Stage::Done` fixture for tests that need a terminal-stage story.
pub(crate) fn done_stage() -> Stage {
Stage::Done {
merged_at: Utc::now(),
merge_commit: GitSha("abc123".to_string()),
}
}