P1 backend: chat persistence, tc-llm providers, runtime loop, gateway SSE

- tc-db: sessions/messages/steps/runs/run_events repos (atomic seq assignment,
  history with ordered step traces, journal replay-from-offset); migration 0003
- tc-llm: provider-neutral ChatRequest/LlmEvent; ScriptedProvider (scenario
  TOML, word-level deltas, multi-turn tool legs — ships in production for
  e2e/air-gap smoke), AnthropicProvider (Messages SSE), OpenAiCompatProvider
  (vLLM/Ollama/llama.cpp); opt-in live tests via TC_LIVE_LLM=1
- tc-runtime: run loop with persist-before-emit event journal, real built-in
  clock.now tool, step rows on the reply message, tool-error resilience,
  broadcast channels for live attach
- tc-api: agent CRUD + settings/full (tenant-isolated, RBAC'd, audited),
  sessions create/list/history?tools=true, POST /api/gateway SSE with
  monotonic ids and exact resumeFrom journal replay (tested equal to live)
- teamclaw-server: config-driven provider factory

83 Rust tests green, all against real Postgres / real TCP.

Co-Authored-By: Claude Fable 5 <[email protected]>
This commit is contained in:
Omar Sobh
2026-06-09 23:16:06 -05:00
co-authored by Claude Fable 5
parent fc173f170d
commit 32008c9ef0
58 changed files with 4378 additions and 11 deletions
@@ -0,0 +1,32 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO sessions (id, agent_id, workspace_id, title, shard)\n VALUES ($1, $2, $3, $4, $5)\n RETURNING created_at, last_active_at",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 1,
"name": "last_active_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid",
"Uuid",
"Uuid",
"Text",
"Int2"
]
},
"nullable": [
false,
false
]
},
"hash": "0c47bc74c67351690bdae097a2b50ace9893c402b22672ccc03a10f44e8f65fd"
}
@@ -0,0 +1,14 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE sessions SET last_active_at = clock_timestamp() WHERE id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": []
},
"hash": "16ad11d77d46c6ebecc3315db05ea02d59f9062356860dd296eda332c5d7bfd6"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO agent_runs (id, session_id, state) VALUES ($1, $2, 'running')",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Uuid"
]
},
"nullable": []
},
"hash": "2a898af5b0fa628fdc353a88511c84b0f1c9aa980c5cb6adac82bf4f6016bde7"
}
@@ -0,0 +1,46 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, session_id, state, last_event_id, error\n FROM agent_runs WHERE session_id = $1\n ORDER BY created_at DESC, id DESC LIMIT 1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "session_id",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "state",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "last_event_id",
"type_info": "Int8"
},
{
"ordinal": 4,
"name": "error",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
false,
false,
false,
true
]
},
"hash": "31433abcda79e8da34cfd7e9ec6caaafe67654c201a161c5f777a98228029a5e"
}
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE agent_runs\n SET state = $2, error = COALESCE($3, error), updated_at = now()\n WHERE id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "35f955dd3e2e74cc3b81f2aaf47dc7d690ee39fe52ff790d71709f84cb347a28"
}
@@ -0,0 +1,22 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO steps\n (id, message_id, seq, kind, tool_name, input, output, taint, status)\n VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Uuid",
"Int4",
"Text",
"Text",
"Jsonb",
"Jsonb",
"TextArray",
"Text"
]
},
"nullable": []
},
"hash": "5253932f240956c50c7a0b215bdab26ec2cfc67a7f0fd4eaf0806f7eea74fa3c"
}
@@ -0,0 +1,31 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO messages (id, session_id, seq, role, content)\n SELECT $1, $2, COALESCE(MAX(seq), 0) + 1, $3, $4\n FROM messages WHERE session_id = $2\n RETURNING seq, created_at",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "seq",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid",
"Uuid",
"Text",
"Jsonb"
]
},
"nullable": [
false,
false
]
},
"hash": "552dd276a90fe534077768e240d4241c6b1b884ebc18f3820104c2551f0d3ad3"
}
@@ -0,0 +1,58 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, agent_id, workspace_id, title, shard, created_at, last_active_at\n FROM sessions WHERE agent_id = $1\n ORDER BY last_active_at DESC, id DESC",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "agent_id",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Uuid"
},
{
"ordinal": 3,
"name": "title",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "shard",
"type_info": "Int2"
},
{
"ordinal": 5,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 6,
"name": "last_active_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false
]
},
"hash": "6e28504ce25e1ba5896c68e39420d9abff368c62d9b5c9cb6600dd3461f8f786"
}
@@ -0,0 +1,46 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, session_id, state, last_event_id, error FROM agent_runs WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "session_id",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "state",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "last_event_id",
"type_info": "Int8"
},
{
"ordinal": 4,
"name": "error",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
false,
false,
false,
true
]
},
"hash": "71fb175f8955ca9b0b70f5ebbb9ca51d7f6c7e02828a8692d477c70737a4f609"
}
@@ -0,0 +1,41 @@
{
"db_name": "PostgreSQL",
"query": "SELECT run_id, seq, event_type, payload\n FROM run_events WHERE run_id = $1 AND seq > $2 ORDER BY seq",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "run_id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "seq",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "event_type",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "payload",
"type_info": "Jsonb"
}
],
"parameters": {
"Left": [
"Uuid",
"Int8"
]
},
"nullable": [
false,
false,
false,
false
]
},
"hash": "920874ea7c9af6145486dfba2baacdbae8af6a9e76f20ab8d40b01981c9ecabc"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE agent_runs SET last_event_id = $2, updated_at = now() WHERE id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Int8"
]
},
"nullable": []
},
"hash": "a74f51eb7192813b02b32df6439bc3aa605d2f25205af0b2ba54b5698b8452cb"
}
@@ -0,0 +1,17 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO run_events (run_id, seq, event_type, payload)\n VALUES ($1, $2, $3, $4)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Int8",
"Text",
"Jsonb"
]
},
"nullable": []
},
"hash": "cc5820b8c6b62ea401c8a97139a2d579e03b2937c41583918387e3692aa7a993"
}
@@ -0,0 +1,58 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, agent_id, workspace_id, title, shard, created_at, last_active_at\n FROM sessions WHERE id = $1",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "agent_id",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "workspace_id",
"type_info": "Uuid"
},
{
"ordinal": 3,
"name": "title",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "shard",
"type_info": "Int2"
},
{
"ordinal": 5,
"name": "created_at",
"type_info": "Timestamptz"
},
{
"ordinal": 6,
"name": "last_active_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false
]
},
"hash": "d8747260733e5801e1f479c525d433cabcda8f10c6c0b1b103d7c49ea1890b9e"
}
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE agents SET\n name = COALESCE($2, name),\n job_title = COALESCE($3, job_title),\n system_prompt = COALESCE($4, system_prompt),\n avatar = COALESCE($5, avatar),\n accent = COALESCE($6, accent),\n wallpaper = COALESCE($7, wallpaper)\n WHERE id = $1 AND deleted_at IS NULL",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Text",
"Text",
"Text",
"Text",
"Text",
"Text"
]
},
"nullable": []
},
"hash": "e147081b726a7e4470f721ead806dd19cb0fa97fbf0b295d155051bd3e6d0480"
}
@@ -0,0 +1,52 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, session_id, seq, role, content, created_at\n FROM messages WHERE session_id = $1 ORDER BY seq",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "session_id",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "seq",
"type_info": "Int8"
},
{
"ordinal": 3,
"name": "role",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "content",
"type_info": "Jsonb"
},
{
"ordinal": 5,
"name": "created_at",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
false,
false,
false,
false,
false
]
},
"hash": "e17d3ecf251b29b1e690cb07b3b0da410ceffae886f5d9379afbf7e95a52e0c2"
}
@@ -0,0 +1,70 @@
{
"db_name": "PostgreSQL",
"query": "SELECT s.id, s.message_id, s.seq, s.kind, s.tool_name, s.input,\n s.output, s.taint, s.status\n FROM steps s\n JOIN messages m ON m.id = s.message_id\n WHERE m.session_id = $1\n ORDER BY m.seq, s.seq",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "message_id",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "seq",
"type_info": "Int4"
},
{
"ordinal": 3,
"name": "kind",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "tool_name",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "input",
"type_info": "Jsonb"
},
{
"ordinal": 6,
"name": "output",
"type_info": "Jsonb"
},
{
"ordinal": 7,
"name": "taint",
"type_info": "TextArray"
},
{
"ordinal": 8,
"name": "status",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
false,
false,
false,
true,
true,
true,
false,
false
]
},
"hash": "e37beba4b1ae596ba5f94ea5a638d45bbfc8e5fdb79d9244272496f540f9e1b9"
}
@@ -0,0 +1,76 @@
{
"db_name": "PostgreSQL",
"query": "SELECT id, workspace_id, name, job_title, system_prompt, avatar,\n accent, wallpaper, managed_by, status\n FROM agents WHERE id = $1 AND deleted_at IS NULL",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Uuid"
},
{
"ordinal": 1,
"name": "workspace_id",
"type_info": "Uuid"
},
{
"ordinal": 2,
"name": "name",
"type_info": "Text"
},
{
"ordinal": 3,
"name": "job_title",
"type_info": "Text"
},
{
"ordinal": 4,
"name": "system_prompt",
"type_info": "Text"
},
{
"ordinal": 5,
"name": "avatar",
"type_info": "Text"
},
{
"ordinal": 6,
"name": "accent",
"type_info": "Text"
},
{
"ordinal": 7,
"name": "wallpaper",
"type_info": "Text"
},
{
"ordinal": 8,
"name": "managed_by",
"type_info": "Uuid"
},
{
"ordinal": 9,
"name": "status",
"type_info": "Text"
}
],
"parameters": {
"Left": [
"Uuid"
]
},
"nullable": [
false,
false,
false,
false,
false,
false,
false,
false,
false,
false
]
},
"hash": "e4d31748aac3f7a3cb6e3e85d8b4c3c45464c2c9f5291341df7e7f4a5e449a16"
}
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE messages SET content = $2 WHERE id = $1",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Uuid",
"Jsonb"
]
},
"nullable": []
},
"hash": "f79fabf13ae2607323542cd133d1b9c98380f5dd0ec47a627e95ba4f7e407a63"
}
Generated
+93
View File
@@ -625,6 +625,17 @@ dependencies = [
"pin-project-lite", "pin-project-lite",
] ]
[[package]]
name = "eventsource-stream"
version = "0.2.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "74fef4569247a5f429d9156b9d0a2599914385dd189c539334c625d8099d90ab"
dependencies = [
"futures-core",
"nom",
"pin-project-lite",
]
[[package]] [[package]]
name = "fastrand" name = "fastrand"
version = "2.4.1" version = "2.4.1"
@@ -1393,6 +1404,12 @@ version = "0.3.17"
source = "registry+https://github.com/rust-lang/crates.io-index" source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
[[package]]
name = "minimal-lexical"
version = "0.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
[[package]] [[package]]
name = "mio" name = "mio"
version = "1.2.1" version = "1.2.1"
@@ -1404,6 +1421,16 @@ dependencies = [
"windows-sys 0.61.2", "windows-sys 0.61.2",
] ]
[[package]]
name = "nom"
version = "7.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d273983c5a657a70a3e8f2a01329822f3b8c8172b73826411a55751e404a0a4a"
dependencies = [
"memchr",
"minimal-lexical",
]
[[package]] [[package]]
name = "num" name = "num"
version = "0.4.3" version = "0.4.3"
@@ -2013,6 +2040,7 @@ dependencies = [
"base64", "base64",
"bytes", "bytes",
"futures-core", "futures-core",
"futures-util",
"http", "http",
"http-body", "http-body",
"http-body-util", "http-body-util",
@@ -2032,12 +2060,14 @@ dependencies = [
"sync_wrapper", "sync_wrapper",
"tokio", "tokio",
"tokio-rustls", "tokio-rustls",
"tokio-util",
"tower", "tower",
"tower-http", "tower-http",
"tower-service", "tower-service",
"url", "url",
"wasm-bindgen", "wasm-bindgen",
"wasm-bindgen-futures", "wasm-bindgen-futures",
"wasm-streams",
"web-sys", "web-sys",
"webpki-roots 1.0.7", "webpki-roots 1.0.7",
] ]
@@ -2733,7 +2763,10 @@ dependencies = [
name = "tc-api" name = "tc-api"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"async-stream",
"axum", "axum",
"eventsource-stream",
"futures",
"reqwest", "reqwest",
"serde", "serde",
"serde_json", "serde_json",
@@ -2741,10 +2774,14 @@ dependencies = [
"tc-auth", "tc-auth",
"tc-db", "tc-db",
"tc-domain", "tc-domain",
"tc-llm",
"tc-runtime",
"tc-testkit", "tc-testkit",
"thiserror", "thiserror",
"time", "time",
"tokio", "tokio",
"urlencoding",
"uuid",
] ]
[[package]] [[package]]
@@ -2799,6 +2836,41 @@ dependencies = [
"uuid", "uuid",
] ]
[[package]]
name = "tc-llm"
version = "0.1.0"
dependencies = [
"async-stream",
"async-trait",
"eventsource-stream",
"futures",
"reqwest",
"serde",
"serde_json",
"thiserror",
"tokio",
"toml",
]
[[package]]
name = "tc-runtime"
version = "0.1.0"
dependencies = [
"async-trait",
"futures",
"serde",
"serde_json",
"sqlx",
"tc-db",
"tc-domain",
"tc-llm",
"tc-testkit",
"thiserror",
"time",
"tokio",
"uuid",
]
[[package]] [[package]]
name = "tc-testkit" name = "tc-testkit"
version = "0.1.0" version = "0.1.0"
@@ -2821,6 +2893,8 @@ dependencies = [
"tc-config", "tc-config",
"tc-db", "tc-db",
"tc-domain", "tc-domain",
"tc-llm",
"tc-runtime",
"time", "time",
"tokio", "tokio",
] ]
@@ -3290,6 +3364,12 @@ dependencies = [
"serde_derive", "serde_derive",
] ]
[[package]]
name = "urlencoding"
version = "2.1.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da"
[[package]] [[package]]
name = "utf8-zero" name = "utf8-zero"
version = "0.8.1" version = "0.8.1"
@@ -3451,6 +3531,19 @@ dependencies = [
"wasmparser", "wasmparser",
] ]
[[package]]
name = "wasm-streams"
version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65"
dependencies = [
"futures-util",
"js-sys",
"wasm-bindgen",
"wasm-bindgen-futures",
"web-sys",
]
[[package]] [[package]]
name = "wasmparser" name = "wasmparser"
version = "0.244.0" version = "0.244.0"
+2
View File
@@ -4,6 +4,8 @@ members = [
"crates/tc-domain", "crates/tc-domain",
"crates/tc-config", "crates/tc-config",
"crates/tc-db", "crates/tc-db",
"crates/tc-llm",
"crates/tc-runtime",
"crates/tc-testkit", "crates/tc-testkit",
"crates/tc-auth", "crates/tc-auth",
"crates/tc-api", "crates/tc-api",
+2
View File
@@ -13,6 +13,8 @@ tc-api = { path = "../../tc-api" }
tc-auth = { path = "../../tc-auth" } tc-auth = { path = "../../tc-auth" }
tc-config = { path = "../../tc-config" } tc-config = { path = "../../tc-config" }
tc-db = { path = "../../tc-db" } tc-db = { path = "../../tc-db" }
tc-llm = { path = "../../tc-llm" }
tc-runtime = { path = "../../tc-runtime" }
tc-domain = { path = "../../tc-domain" } tc-domain = { path = "../../tc-domain" }
time = { workspace = true } time = { workspace = true }
tokio = { workspace = true } tokio = { workspace = true }
+47 -5
View File
@@ -1,12 +1,15 @@
//! TeamClaw server: REST API, and (in later phases) the streaming gateway, //! TeamClaw server: REST API and streaming gateway (later phases add the
//! scheduler, and safety worker, composed into one binary. //! scheduler and safety worker) composed into one binary.
mod e2e; mod e2e;
use std::path::PathBuf; use std::path::{Path, PathBuf};
use std::process::ExitCode; use std::process::ExitCode;
use std::sync::Arc;
use tc_config::AppConfig; use tc_config::{AppConfig, LlmProviderKind};
use tc_llm::{AnthropicProvider, LlmProvider, OpenAiCompatProvider, ScriptedProvider};
use tc_runtime::{Runtime, RuntimeConfig};
#[tokio::main] #[tokio::main]
async fn main() -> ExitCode { async fn main() -> ExitCode {
@@ -19,6 +22,35 @@ async fn main() -> ExitCode {
} }
} }
/// Instantiates the configured LLM provider. The Anthropic key comes from
/// the environment until the secret broker lands in P2.
fn build_provider(config: &AppConfig) -> Result<Arc<dyn LlmProvider>, String> {
match config.llm.provider {
LlmProviderKind::Anthropic => {
let key = std::env::var("ANTHROPIC_API_KEY")
.map_err(|_| "llm.provider = \"anthropic\" requires ANTHROPIC_API_KEY")?;
Ok(Arc::new(AnthropicProvider::new(key)))
}
LlmProviderKind::OpenAiCompat => {
let base_url = config.llm.base_url.clone().expect("validated by tc-config");
Ok(Arc::new(OpenAiCompatProvider::new(
base_url,
std::env::var("TEAMCLAW_LLM_API_KEY").ok(),
)))
}
LlmProviderKind::Scripted => {
let path = config
.llm
.scenario_path
.clone()
.expect("validated by tc-config");
let provider = ScriptedProvider::from_path(Path::new(&path))
.map_err(|e| format!("scenario load failed: {e}"))?;
Ok(Arc::new(provider))
}
}
}
async fn run() -> Result<(), String> { async fn run() -> Result<(), String> {
let config_path = let config_path =
PathBuf::from(std::env::var("TEAMCLAW_CONFIG").unwrap_or_else(|_| "teamclaw.toml".into())); PathBuf::from(std::env::var("TEAMCLAW_CONFIG").unwrap_or_else(|_| "teamclaw.toml".into()));
@@ -36,7 +68,17 @@ async fn run() -> Result<(), String> {
e2e::seed(&pool).await?; e2e::seed(&pool).await?;
} }
let app = tc_api::router(tc_api::AppState::new(pool)); let provider = build_provider(&config)?;
let runtime = Runtime::new(
pool.clone(),
provider,
RuntimeConfig {
model: config.llm.model.clone(),
max_tokens: 4096,
},
);
let app = tc_api::router(tc_api::AppState::new(pool, runtime));
let listener = tokio::net::TcpListener::bind(config.listen_addr) let listener = tokio::net::TcpListener::bind(config.listen_addr)
.await .await
.map_err(|e| format!("bind {} failed: {e}", config.listen_addr))?; .map_err(|e| format!("bind {} failed: {e}", config.listen_addr))?;
+8
View File
@@ -7,23 +7,31 @@ license.workspace = true
publish.workspace = true publish.workspace = true
[dependencies] [dependencies]
async-stream = "0.3"
axum = "0.8" axum = "0.8"
futures = "0.3"
serde = { workspace = true } serde = { workspace = true }
serde_json = { workspace = true } serde_json = { workspace = true }
sqlx = { workspace = true } sqlx = { workspace = true }
tc-auth = { path = "../tc-auth" } tc-auth = { path = "../tc-auth" }
tc-db = { path = "../tc-db" } tc-db = { path = "../tc-db" }
tc-domain = { path = "../tc-domain" } tc-domain = { path = "../tc-domain" }
tc-runtime = { path = "../tc-runtime" }
thiserror = { workspace = true } thiserror = { workspace = true }
time = { workspace = true } time = { workspace = true }
tokio = { workspace = true } tokio = { workspace = true }
uuid = { workspace = true }
[dev-dependencies] [dev-dependencies]
eventsource-stream = "0.2"
reqwest = { version = "0.12", default-features = false, features = [ reqwest = { version = "0.12", default-features = false, features = [
"json", "json",
"rustls-tls", "rustls-tls",
"stream",
] } ] }
tc-llm = { path = "../tc-llm" }
tc-testkit = { path = "../tc-testkit" } tc-testkit = { path = "../tc-testkit" }
urlencoding = "2"
[lints] [lints]
workspace = true workspace = true
+20 -3
View File
@@ -4,10 +4,11 @@ mod error;
mod extract; mod extract;
mod routes; mod routes;
use axum::routing::{get, post}; use axum::routing::{delete, get, patch, post};
use axum::Router; use axum::Router;
use sqlx::PgPool; use sqlx::PgPool;
use tc_auth::AuthService; use tc_auth::AuthService;
use tc_runtime::Runtime;
pub use error::ApiError; pub use error::ApiError;
pub use extract::Authed; pub use extract::Authed;
@@ -16,12 +17,17 @@ pub use extract::Authed;
pub struct AppState { pub struct AppState {
pub pool: PgPool, pub pool: PgPool,
pub auth: AuthService, pub auth: AuthService,
pub runtime: Runtime,
} }
impl AppState { impl AppState {
pub fn new(pool: PgPool) -> AppState { pub fn new(pool: PgPool, runtime: Runtime) -> AppState {
let auth = AuthService::new(pool.clone()); let auth = AuthService::new(pool.clone());
AppState { pool, auth } AppState {
pool,
auth,
runtime,
}
} }
} }
@@ -31,6 +37,17 @@ pub fn router(state: AppState) -> Router {
.route("/api/auth/login", post(routes::auth::login)) .route("/api/auth/login", post(routes::auth::login))
.route("/api/auth/logout", post(routes::auth::logout)) .route("/api/auth/logout", post(routes::auth::logout))
.route("/api/user/me", get(routes::identity::me)) .route("/api/user/me", get(routes::identity::me))
.route("/api/claws", post(routes::claws::create))
.route("/api/claws/{id}", patch(routes::claws::patch))
.route("/api/claws/{id}", delete(routes::claws::delete))
.route(
"/api/claws/settings/full",
get(routes::claws::settings_full),
)
.route("/api/sessions", get(routes::sessions::list))
.route("/api/sessions", post(routes::sessions::create))
.route("/api/sessions/history", get(routes::sessions::history))
.route("/api/gateway", post(routes::gateway::gateway))
.route("/api/team/claws", get(routes::team::claws)) .route("/api/team/claws", get(routes::team::claws))
.route("/api/team/members", get(routes::team::members)) .route("/api/team/members", get(routes::team::members))
.route("/api/team/credits", get(routes::team::credits)) .route("/api/team/credits", get(routes::team::credits))
+158
View File
@@ -0,0 +1,158 @@
use axum::extract::{Path, Query, State};
use axum::http::StatusCode;
use axum::Json;
use serde::Deserialize;
use serde_json::{json, Value};
use tc_db::repo::audit::Actor;
use tc_domain::{AccessPolicy, Agent, AgentId, AgentStatus};
use crate::{ApiError, AppState, Authed};
/// Loads an agent and enforces tenant isolation: agents in other workspaces
/// are indistinguishable from non-existent ones.
pub(crate) async fn workspace_agent(
state: &AppState,
user: &tc_auth::AuthedUser,
agent_id: AgentId,
) -> Result<Agent, ApiError> {
let agent = tc_db::repo::agents::get(&state.pool, agent_id).await?;
if agent.workspace_id != user.workspace_id {
return Err(ApiError::NotFound);
}
Ok(agent)
}
#[derive(Deserialize)]
pub struct CreateClawRequest {
name: String,
job_title: String,
#[serde(default)]
system_prompt: String,
#[serde(default)]
avatar: String,
#[serde(default)]
accent: String,
#[serde(default)]
wallpaper: String,
}
/// POST /api/claws — completing creation yields a LIVE agent (§9).
pub async fn create(
State(state): State<AppState>,
Authed(user): Authed,
Json(body): Json<CreateClawRequest>,
) -> Result<(StatusCode, Json<Agent>), ApiError> {
let agent = Agent {
id: AgentId::new(),
workspace_id: user.workspace_id,
name: body.name,
job_title: body.job_title,
system_prompt: body.system_prompt,
avatar: body.avatar,
accent: body.accent,
wallpaper: body.wallpaper,
managed_by: user.user_id,
status: AgentStatus::Online,
};
tc_db::repo::agents::insert(&state.pool, &agent, &AccessPolicy::default()).await?;
tc_db::repo::audit::append(
&state.pool,
user.workspace_id,
Actor::User(user.user_id),
"agent.created",
"agent",
&agent.id.to_string(),
json!({"name": agent.name, "job_title": agent.job_title}),
)
.await?;
Ok((StatusCode::CREATED, Json(agent)))
}
#[derive(Deserialize)]
pub struct PatchClawRequest {
name: Option<String>,
job_title: Option<String>,
system_prompt: Option<String>,
avatar: Option<String>,
accent: Option<String>,
wallpaper: Option<String>,
}
/// PATCH /api/claws/{id} — Edit profile (§7.7).
pub async fn patch(
State(state): State<AppState>,
Authed(user): Authed,
Path(id): Path<AgentId>,
Json(body): Json<PatchClawRequest>,
) -> Result<Json<Agent>, ApiError> {
workspace_agent(&state, &user, id).await?;
let updated = tc_db::repo::agents::update_profile(
&state.pool,
id,
body.name.as_deref(),
body.job_title.as_deref(),
body.system_prompt.as_deref(),
body.avatar.as_deref(),
body.accent.as_deref(),
body.wallpaper.as_deref(),
)
.await?;
tc_db::repo::audit::append(
&state.pool,
user.workspace_id,
Actor::User(user.user_id),
"agent.updated",
"agent",
&id.to_string(),
json!({}),
)
.await?;
Ok(Json(updated))
}
/// DELETE /api/claws/{id} — destructive (§7.7): workspace owners or the
/// claw's manager only. Soft delete keeps rows for audit.
pub async fn delete(
State(state): State<AppState>,
Authed(user): Authed,
Path(id): Path<AgentId>,
) -> Result<StatusCode, ApiError> {
let agent = workspace_agent(&state, &user, id).await?;
if !user.role.is_owner() && agent.managed_by != user.user_id {
return Err(ApiError::Forbidden);
}
tc_db::repo::agents::soft_delete(&state.pool, id).await?;
tc_db::repo::audit::append(
&state.pool,
user.workspace_id,
Actor::User(user.user_id),
"agent.deleted",
"agent",
&id.to_string(),
json!({"name": agent.name}),
)
.await?;
Ok(StatusCode::NO_CONTENT)
}
#[derive(Deserialize)]
pub struct SettingsQuery {
#[serde(rename = "clawId")]
claw_id: AgentId,
}
/// GET /api/claws/settings/full?clawId= — Settings panel aggregate (§7.7).
pub async fn settings_full(
State(state): State<AppState>,
Authed(user): Authed,
Query(query): Query<SettingsQuery>,
) -> Result<Json<Value>, ApiError> {
let agent = workspace_agent(&state, &user, query.claw_id).await?;
let policy = tc_db::repo::agents::access_policy(&state.pool, agent.id).await?;
let manager = tc_db::repo::users::get(&state.pool, agent.managed_by).await?;
Ok(Json(json!({
"agent": agent,
"access_policy": policy,
"managed_by_name": manager.display_name,
})))
}
+140
View File
@@ -0,0 +1,140 @@
use std::convert::Infallible;
use std::time::Duration;
use axum::extract::{Query, State};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::Json;
use futures::stream::BoxStream;
use serde::Deserialize;
use tc_domain::AgentId;
use tc_runtime::{RunEventBody, RunEventEnvelope};
use tokio::sync::broadcast;
use crate::routes::claws::workspace_agent;
use crate::routes::sessions::scoped_session;
use crate::{ApiError, AppState, Authed};
#[derive(Deserialize)]
pub struct GatewayQuery {
#[serde(rename = "clawId")]
claw_id: AgentId,
}
#[derive(Deserialize)]
pub struct GatewayRequest {
#[serde(rename = "sessionKey")]
session_key: String,
/// New user message — starts a run.
message: Option<String>,
/// Reconnect offset — replays the journal after this seq, then goes
/// live if the run is still streaming.
#[serde(rename = "resumeFrom")]
resume_from: Option<i64>,
}
fn envelope_to_sse(envelope: &RunEventEnvelope) -> Event {
let payload = serde_json::to_value(&envelope.event).expect("event serializes");
Event::default()
.id(envelope.seq.to_string())
.event(payload["type"].as_str().expect("tagged event"))
.data(payload.to_string())
}
fn is_terminal(event: &RunEventBody) -> bool {
matches!(
event,
RunEventBody::RunCompleted { .. } | RunEventBody::Error { .. }
)
}
type SseStream = BoxStream<'static, Result<Event, Infallible>>;
fn live_stream(mut rx: broadcast::Receiver<RunEventEnvelope>, after: i64) -> SseStream {
Box::pin(async_stream::stream! {
loop {
match rx.recv().await {
Ok(envelope) => {
if envelope.seq <= after {
continue;
}
let done = is_terminal(&envelope.event);
yield Ok(envelope_to_sse(&envelope));
if done {
break;
}
}
Err(broadcast::error::RecvError::Lagged(_)) => continue,
Err(broadcast::error::RecvError::Closed) => break,
}
}
})
}
/// POST /api/gateway?clawId= — the single audited streaming channel (§15).
/// Every emitted event was journaled first, so a reconnect with
/// `resumeFrom` replays exactly what live observers saw.
pub async fn gateway(
State(state): State<AppState>,
Authed(user): Authed,
Query(query): Query<GatewayQuery>,
Json(body): Json<GatewayRequest>,
) -> Result<impl axum::response::IntoResponse, ApiError> {
let agent = workspace_agent(&state, &user, query.claw_id).await?;
let session = scoped_session(&state, &user, &body.session_key).await?;
if session.agent_id != agent.id {
return Err(ApiError::NotFound);
}
let stream: SseStream = match body.message {
Some(text) => {
let started = state
.runtime
.send_message(session.id, &text)
.await
.map_err(|e| match e {
tc_runtime::RuntimeError::Db(tc_db::DbError::NotFound) => ApiError::NotFound,
_ => ApiError::Internal,
})?;
live_stream(started.events, 0)
}
None => {
let resume_from = body.resume_from.unwrap_or(0);
let run = tc_db::repo::runs::latest_for_session(&state.pool, session.id)
.await?
.ok_or(ApiError::NotFound)?;
// Subscribe before reading the journal so no event falls in the
// gap; the live tail then skips anything the replay covered.
let live = state.runtime.subscribe(run.id).await;
let journal =
tc_db::repo::run_events::list_after(&state.pool, run.id, resume_from).await?;
let last_replayed = journal.last().map(|e| e.seq).unwrap_or(resume_from);
let replay_done = journal
.last()
.map(|e| e.event_type == "run_completed" || e.event_type == "error")
.unwrap_or(false);
Box::pin(async_stream::stream! {
for entry in journal {
yield Ok(Event::default()
.id(entry.seq.to_string())
.event(entry.event_type.clone())
.data(entry.payload.to_string()));
}
if !replay_done {
if let Some(rx) = live {
let mut tail = live_stream(rx, last_replayed);
while let Some(event) = futures::StreamExt::next(&mut tail).await {
yield event;
}
}
}
})
}
};
Ok(Sse::new(stream).keep_alive(
KeepAlive::new()
.interval(Duration::from_secs(15))
.text("hb"),
))
}
+3
View File
@@ -1,4 +1,7 @@
pub mod auth; pub mod auth;
pub mod claws;
pub mod gateway;
pub mod health; pub mod health;
pub mod identity; pub mod identity;
pub mod sessions;
pub mod team; pub mod team;
+118
View File
@@ -0,0 +1,118 @@
use std::str::FromStr;
use axum::extract::{Query, State};
use axum::http::StatusCode;
use axum::Json;
use serde::Deserialize;
use serde_json::{json, Value};
use tc_domain::{AgentId, MessageId, MessageRole, MessageWithSteps, Session, SessionKey};
use uuid::Uuid;
use crate::routes::claws::workspace_agent;
use crate::{ApiError, AppState, Authed};
/// Builds the deep-linkable key for a session (§12). The message anchor is
/// the latest message, or the nil uuid for a fresh session.
fn session_key(session: &Session, last_message: Option<MessageId>) -> SessionKey {
SessionKey {
agent_id: session.agent_id,
shard: session.shard,
session_id: session.id,
message_id: last_message.unwrap_or_else(|| MessageId::from(Uuid::nil())),
}
}
fn with_key(session: Session, last_message: Option<MessageId>) -> Value {
let key = session_key(&session, last_message).to_string();
let mut value = serde_json::to_value(&session).expect("session serializes");
value["sessionKey"] = json!(key);
value
}
/// Resolves a sessionKey to its session, enforcing workspace and claw
/// scoping. Used by history and the gateway.
pub(crate) async fn scoped_session(
state: &AppState,
user: &tc_auth::AuthedUser,
raw_key: &str,
) -> Result<Session, ApiError> {
let key = SessionKey::from_str(raw_key).map_err(|_| ApiError::NotFound)?;
let session = tc_db::repo::sessions::get(&state.pool, key.session_id).await?;
if session.agent_id != key.agent_id {
return Err(ApiError::NotFound);
}
workspace_agent(state, user, session.agent_id).await?;
Ok(session)
}
#[derive(Deserialize)]
pub struct ListQuery {
#[serde(rename = "clawId")]
claw_id: AgentId,
}
/// GET /api/sessions?clawId= — sessions column data (§6).
pub async fn list(
State(state): State<AppState>,
Authed(user): Authed,
Query(query): Query<ListQuery>,
) -> Result<Json<Vec<Value>>, ApiError> {
workspace_agent(&state, &user, query.claw_id).await?;
let sessions = tc_db::repo::sessions::list_by_agent(&state.pool, query.claw_id).await?;
Ok(Json(
sessions.into_iter().map(|s| with_key(s, None)).collect(),
))
}
#[derive(Deserialize)]
pub struct CreateRequest {
#[serde(rename = "clawId")]
claw_id: AgentId,
#[serde(default)]
title: String,
}
/// POST /api/sessions — new resumable session (§6 "+ New session").
pub async fn create(
State(state): State<AppState>,
Authed(user): Authed,
Json(body): Json<CreateRequest>,
) -> Result<(StatusCode, Json<Value>), ApiError> {
let agent = workspace_agent(&state, &user, body.claw_id).await?;
let session =
tc_db::repo::sessions::create(&state.pool, agent.id, agent.workspace_id, &body.title)
.await?;
Ok((StatusCode::CREATED, Json(with_key(session, None))))
}
#[derive(Deserialize)]
pub struct HistoryQuery {
#[serde(rename = "sessionKey")]
session_key: String,
#[serde(default)]
tools: bool,
}
/// GET /api/sessions/history?sessionKey=&tools=true — the durable
/// transcript; identical data to what the gateway streamed (§13).
pub async fn history(
State(state): State<AppState>,
Authed(user): Authed,
Query(query): Query<HistoryQuery>,
) -> Result<Json<Vec<MessageWithSteps>>, ApiError> {
let session = scoped_session(&state, &user, &query.session_key).await?;
let mut history = tc_db::repo::messages::history(&state.pool, session.id).await?;
if !query.tools {
for entry in &mut history {
entry.steps.clear();
}
}
// The empty in-flight reply row is an implementation detail; history
// consumers only see finalized or non-empty messages.
history.retain(|entry| {
entry.message.role != MessageRole::Agent
|| entry.message.content["text"].as_str() != Some("")
|| !entry.steps.is_empty()
});
Ok(Json(history))
}
+254
View File
@@ -0,0 +1,254 @@
use serde_json::{json, Value};
use tc_api::AppState;
use tc_auth::AuthService;
use tc_domain::{Role, User, UserId, Workspace, WorkspaceId};
struct TestServer {
base: String,
client: reqwest::Client,
}
fn test_runtime(pool: sqlx::PgPool) -> tc_runtime::Runtime {
tc_runtime::Runtime::new(
pool,
std::sync::Arc::new(tc_llm::ScriptedProvider::from_toml("").unwrap()),
tc_runtime::RuntimeConfig {
model: "scripted".into(),
max_tokens: 1024,
},
)
}
async fn serve(pool: sqlx::PgPool) -> TestServer {
let app = tc_api::router(AppState::new(pool.clone(), test_runtime(pool.clone())));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
TestServer {
base: format!("http://{addr}"),
client: reqwest::Client::new(),
}
}
async fn seed_user(pool: &sqlx::PgPool, ws: &Workspace, email: &str, role: Role) -> User {
let user = User {
id: UserId::new(),
workspace_id: ws.id,
email: email.into(),
role,
display_name: email.split('@').next().unwrap().into(),
created_at: time::OffsetDateTime::UNIX_EPOCH,
};
tc_db::repo::users::insert(pool, &user).await.unwrap();
AuthService::new(pool.clone())
.set_password(user.id, "pw")
.await
.unwrap();
user
}
async fn seed_workspace(pool: &sqlx::PgPool) -> Workspace {
let ws = Workspace {
id: WorkspaceId::new(),
name: "Acme".into(),
plan: "team".into(),
};
tc_db::repo::workspaces::insert(pool, &ws).await.unwrap();
ws
}
async fn login(server: &TestServer, email: &str) -> String {
let res = server
.client
.post(format!("{}/api/auth/login", server.base))
.json(&json!({"email": email, "password": "pw"}))
.send()
.await
.unwrap();
assert_eq!(res.status(), 200);
res.json::<Value>().await.unwrap()["token"]
.as_str()
.unwrap()
.to_owned()
}
async fn create_claw(server: &TestServer, token: &str, name: &str) -> Value {
let res = server
.client
.post(format!("{}/api/claws", server.base))
.bearer_auth(token)
.json(&json!({"name": name, "job_title": "Research Analyst"}))
.send()
.await
.unwrap();
assert_eq!(res.status(), 201);
res.json().await.unwrap()
}
#[tokio::test]
async fn create_returns_live_agent_with_default_policy_and_audit() {
let pool = tc_testkit::test_pool().await;
let ws = seed_workspace(&pool).await;
let owner = seed_user(&pool, &ws, "[email protected]", Role::Owner).await;
let server = serve(pool.clone()).await;
let token = login(&server, "[email protected]").await;
let claw = create_claw(&server, &token, "Scout").await;
assert_eq!(claw["name"], "Scout");
assert_eq!(claw["status"], "online");
assert_eq!(claw["managed_by"], owner.id.to_string());
let settings: Value = server
.client
.get(format!(
"{}/api/claws/settings/full?clawId={}",
server.base,
claw["id"].as_str().unwrap()
))
.bearer_auth(&token)
.send()
.await
.unwrap()
.json()
.await
.unwrap();
assert_eq!(settings["agent"]["name"], "Scout");
assert_eq!(settings["access_policy"]["humans"]["mode"], "entire_team");
assert_eq!(settings["access_policy"]["agents"]["mode"], "any");
let audited = sqlx::query_scalar::<_, i64>(
"SELECT count(*) FROM audit_log WHERE event_type = 'agent.created'",
)
.fetch_one(&pool)
.await
.unwrap();
assert_eq!(audited, 1);
}
#[tokio::test]
async fn patch_updates_profile_and_system_prompt() {
let pool = tc_testkit::test_pool().await;
let ws = seed_workspace(&pool).await;
seed_user(&pool, &ws, "[email protected]", Role::Owner).await;
let server = serve(pool).await;
let token = login(&server, "[email protected]").await;
let claw = create_claw(&server, &token, "Scout").await;
let id = claw["id"].as_str().unwrap();
let res = server
.client
.patch(format!("{}/api/claws/{id}", server.base))
.bearer_auth(&token)
.json(&json!({
"job_title": "Senior Analyst",
"system_prompt": "Think in bullet points."
}))
.send()
.await
.unwrap();
assert_eq!(res.status(), 200);
let updated: Value = res.json().await.unwrap();
assert_eq!(updated["job_title"], "Senior Analyst");
assert_eq!(updated["system_prompt"], "Think in bullet points.");
// Untouched fields persist.
assert_eq!(updated["name"], "Scout");
}
#[tokio::test]
async fn delete_requires_owner_or_manager() {
let pool = tc_testkit::test_pool().await;
let ws = seed_workspace(&pool).await;
seed_user(&pool, &ws, "[email protected]", Role::Owner).await;
seed_user(&pool, &ws, "[email protected]", Role::Member).await;
let server = serve(pool).await;
let owner_token = login(&server, "[email protected]").await;
let member_token = login(&server, "[email protected]").await;
let claw = create_claw(&server, &owner_token, "Scout").await;
let id = claw["id"].as_str().unwrap();
// A plain member who doesn't manage the claw cannot delete it.
let forbidden = server
.client
.delete(format!("{}/api/claws/{id}", server.base))
.bearer_auth(&member_token)
.send()
.await
.unwrap();
assert_eq!(forbidden.status(), 403);
let deleted = server
.client
.delete(format!("{}/api/claws/{id}", server.base))
.bearer_auth(&owner_token)
.send()
.await
.unwrap();
assert_eq!(deleted.status(), 204);
// Gone from the roster.
let claws: Value = server
.client
.get(format!("{}/api/team/claws", server.base))
.bearer_auth(&owner_token)
.send()
.await
.unwrap()
.json()
.await
.unwrap();
assert_eq!(claws.as_array().unwrap().len(), 0);
}
#[tokio::test]
async fn members_can_create_and_manage_their_own_claws() {
let pool = tc_testkit::test_pool().await;
let ws = seed_workspace(&pool).await;
seed_user(&pool, &ws, "[email protected]", Role::Member).await;
let server = serve(pool).await;
let token = login(&server, "[email protected]").await;
// Builders (plain members) can create claws (§1 personas)...
let claw = create_claw(&server, &token, "Drafter").await;
let id = claw["id"].as_str().unwrap();
// ...and delete the ones they manage.
let deleted = server
.client
.delete(format!("{}/api/claws/{id}", server.base))
.bearer_auth(&token)
.send()
.await
.unwrap();
assert_eq!(deleted.status(), 204);
}
#[tokio::test]
async fn cross_workspace_access_is_not_found() {
let pool = tc_testkit::test_pool().await;
let ws_a = seed_workspace(&pool).await;
let ws_b = seed_workspace(&pool).await;
seed_user(&pool, &ws_a, "[email protected]", Role::Owner).await;
seed_user(&pool, &ws_b, "[email protected]", Role::Owner).await;
let server = serve(pool).await;
let token_a = login(&server, "[email protected]").await;
let token_b = login(&server, "[email protected]").await;
let claw = create_claw(&server, &token_a, "Scout").await;
let id = claw["id"].as_str().unwrap();
// Tenant isolation: the other workspace can't even see it exists.
let other = server
.client
.get(format!(
"{}/api/claws/settings/full?clawId={id}",
server.base
))
.bearer_auth(&token_b)
.send()
.await
.unwrap();
assert_eq!(other.status(), 404);
}
+326
View File
@@ -0,0 +1,326 @@
use std::str::FromStr;
use std::sync::Arc;
use eventsource_stream::Eventsource;
use futures::StreamExt;
use serde_json::{json, Value};
use tc_api::AppState;
use tc_auth::AuthService;
use tc_domain::{Role, SessionKey, User, UserId, Workspace, WorkspaceId};
use tc_llm::ScriptedProvider;
use tc_runtime::{Runtime, RuntimeConfig};
const SCENARIOS: &str = r#"
[[scenario]]
marker = "[[scenario:tool-time]]"
[[scenario.turns]]
events = [
{ type = "text", text = "Let me check the clock." },
{ type = "tool_use", name = "clock.now", input = {} },
]
[[scenario.turns]]
events = [
{ type = "text", text = " All done." },
]
"#;
struct TestServer {
base: String,
client: reqwest::Client,
}
async fn serve(pool: sqlx::PgPool) -> TestServer {
let runtime = Runtime::new(
pool.clone(),
Arc::new(ScriptedProvider::from_toml(SCENARIOS).unwrap()),
RuntimeConfig {
model: "scripted".into(),
max_tokens: 1024,
},
);
let app = tc_api::router(AppState::new(pool, runtime));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
TestServer {
base: format!("http://{addr}"),
client: reqwest::Client::new(),
}
}
async fn seed_and_login(pool: &sqlx::PgPool, server: &TestServer) -> (String, String) {
let ws = Workspace {
id: WorkspaceId::new(),
name: "Acme".into(),
plan: "team".into(),
};
tc_db::repo::workspaces::insert(pool, &ws).await.unwrap();
let owner = User {
id: UserId::new(),
workspace_id: ws.id,
email: format!("{}@acme.test", UserId::new()),
role: Role::Owner,
display_name: "Owner".into(),
created_at: time::OffsetDateTime::UNIX_EPOCH,
};
tc_db::repo::users::insert(pool, &owner).await.unwrap();
AuthService::new(pool.clone())
.set_password(owner.id, "pw")
.await
.unwrap();
let token = server
.client
.post(format!("{}/api/auth/login", server.base))
.json(&json!({"email": owner.email, "password": "pw"}))
.send()
.await
.unwrap()
.json::<Value>()
.await
.unwrap()["token"]
.as_str()
.unwrap()
.to_owned();
let claw: Value = server
.client
.post(format!("{}/api/claws", server.base))
.bearer_auth(&token)
.json(&json!({"name": "Scout", "job_title": "Analyst"}))
.send()
.await
.unwrap()
.json()
.await
.unwrap();
(token, claw["id"].as_str().unwrap().to_owned())
}
async fn create_session(server: &TestServer, token: &str, claw_id: &str) -> Value {
let res = server
.client
.post(format!("{}/api/sessions", server.base))
.bearer_auth(token)
.json(&json!({"clawId": claw_id, "title": "Research"}))
.send()
.await
.unwrap();
assert_eq!(res.status(), 201);
res.json().await.unwrap()
}
/// Collects SSE events (id, event-name, parsed data) until run completion.
async fn collect_sse(response: reqwest::Response) -> Vec<(String, String, Value)> {
let mut events = Vec::new();
let mut stream = response.bytes_stream().eventsource();
while let Some(event) = stream.next().await {
let event = event.unwrap();
let data: Value = serde_json::from_str(&event.data).unwrap();
let name = event.event.clone();
let done = name == "run_completed" || name == "error";
events.push((event.id, name, data));
if done {
break;
}
}
events
}
#[tokio::test]
async fn session_create_returns_a_parseable_session_key() {
let pool = tc_testkit::test_pool().await;
let server = serve(pool.clone()).await;
let (token, claw_id) = seed_and_login(&pool, &server).await;
let session = create_session(&server, &token, &claw_id).await;
assert_eq!(session["title"], "Research");
let key = SessionKey::from_str(session["sessionKey"].as_str().unwrap()).unwrap();
assert_eq!(key.agent_id.to_string(), claw_id);
}
#[tokio::test]
async fn sessions_list_is_scoped_to_the_claw() {
let pool = tc_testkit::test_pool().await;
let server = serve(pool.clone()).await;
let (token, claw_id) = seed_and_login(&pool, &server).await;
create_session(&server, &token, &claw_id).await;
create_session(&server, &token, &claw_id).await;
let sessions: Value = server
.client
.get(format!("{}/api/sessions?clawId={claw_id}", server.base))
.bearer_auth(&token)
.send()
.await
.unwrap()
.json()
.await
.unwrap();
assert_eq!(sessions.as_array().unwrap().len(), 2);
}
#[tokio::test]
async fn gateway_streams_a_full_run_with_monotonic_ids() {
let pool = tc_testkit::test_pool().await;
let server = serve(pool.clone()).await;
let (token, claw_id) = seed_and_login(&pool, &server).await;
let session = create_session(&server, &token, &claw_id).await;
let session_key = session["sessionKey"].as_str().unwrap();
let response = server
.client
.post(format!("{}/api/gateway?clawId={claw_id}", server.base))
.bearer_auth(&token)
.json(&json!({"sessionKey": session_key, "message": "hello"}))
.send()
.await
.unwrap();
assert_eq!(response.status(), 200);
assert!(response
.headers()
.get("content-type")
.unwrap()
.to_str()
.unwrap()
.starts_with("text/event-stream"));
let events = collect_sse(response).await;
assert_eq!(events[0].1, "run_started");
assert_eq!(events.last().unwrap().1, "run_completed");
let ids: Vec<i64> = events
.iter()
.map(|(id, _, _)| id.parse().unwrap())
.collect();
let mut sorted = ids.clone();
sorted.sort_unstable();
assert_eq!(ids, sorted, "ids must be monotonic");
let text: String = events
.iter()
.filter(|(_, name, _)| name == "text_delta")
.map(|(_, _, data)| data["delta"].as_str().unwrap())
.collect();
assert_eq!(text, "I received: hello");
}
#[tokio::test]
async fn gateway_runs_tools_and_history_replays_with_steps() {
let pool = tc_testkit::test_pool().await;
let server = serve(pool.clone()).await;
let (token, claw_id) = seed_and_login(&pool, &server).await;
let session = create_session(&server, &token, &claw_id).await;
let session_key = session["sessionKey"].as_str().unwrap();
let response = server
.client
.post(format!("{}/api/gateway?clawId={claw_id}", server.base))
.bearer_auth(&token)
.json(&json!({
"sessionKey": session_key,
"message": "time? [[scenario:tool-time]]"
}))
.send()
.await
.unwrap();
let events = collect_sse(response).await;
assert!(events.iter().any(|(_, name, _)| name == "step_started"));
assert!(events
.iter()
.any(|(_, name, data)| { name == "step_finished" && data["status"] == "ok" }));
let history: Value = server
.client
.get(format!(
"{}/api/sessions/history?sessionKey={}&tools=true",
server.base,
urlencoding::encode(session_key)
))
.bearer_auth(&token)
.send()
.await
.unwrap()
.json()
.await
.unwrap();
let messages = history.as_array().unwrap();
assert_eq!(messages.len(), 2);
assert_eq!(messages[1]["role"], "agent");
assert_eq!(messages[1]["steps"][0]["tool_name"], "clock.now");
}
#[tokio::test]
async fn resume_replays_the_exact_journal() {
let pool = tc_testkit::test_pool().await;
let server = serve(pool.clone()).await;
let (token, claw_id) = seed_and_login(&pool, &server).await;
let session = create_session(&server, &token, &claw_id).await;
let session_key = session["sessionKey"].as_str().unwrap();
let live = collect_sse(
server
.client
.post(format!("{}/api/gateway?clawId={claw_id}", server.base))
.bearer_auth(&token)
.json(&json!({"sessionKey": session_key, "message": "ping"}))
.send()
.await
.unwrap(),
)
.await;
// Reconnect with resumeFrom=0: the same events come back from the
// journal, byte-for-byte equal data.
let replayed = collect_sse(
server
.client
.post(format!("{}/api/gateway?clawId={claw_id}", server.base))
.bearer_auth(&token)
.json(&json!({"sessionKey": session_key, "resumeFrom": 0}))
.send()
.await
.unwrap(),
)
.await;
assert_eq!(live, replayed);
// Partial resume skips already-seen events.
let tail = collect_sse(
server
.client
.post(format!("{}/api/gateway?clawId={claw_id}", server.base))
.bearer_auth(&token)
.json(&json!({"sessionKey": session_key, "resumeFrom": live[1].0.parse::<i64>().unwrap()}))
.send()
.await
.unwrap(),
)
.await;
assert_eq!(tail.len(), live.len() - 2);
}
#[tokio::test]
async fn gateway_rejects_claws_outside_the_workspace() {
let pool = tc_testkit::test_pool().await;
let server = serve(pool.clone()).await;
let (token, claw_id) = seed_and_login(&pool, &server).await;
let (other_token, _) = seed_and_login(&pool, &server).await;
let session = create_session(&server, &token, &claw_id).await;
let response = server
.client
.post(format!("{}/api/gateway?clawId={claw_id}", server.base))
.bearer_auth(&other_token)
.json(&json!({
"sessionKey": session["sessionKey"],
"message": "hi"
}))
.send()
.await
.unwrap();
assert_eq!(response.status(), 404);
}
+12 -1
View File
@@ -11,8 +11,19 @@ struct TestServer {
} }
/// Boots the real axum server on an ephemeral port over real TCP. /// Boots the real axum server on an ephemeral port over real TCP.
fn test_runtime(pool: sqlx::PgPool) -> tc_runtime::Runtime {
tc_runtime::Runtime::new(
pool,
std::sync::Arc::new(tc_llm::ScriptedProvider::from_toml("").unwrap()),
tc_runtime::RuntimeConfig {
model: "scripted".into(),
max_tokens: 1024,
},
)
}
async fn serve(pool: sqlx::PgPool) -> TestServer { async fn serve(pool: sqlx::PgPool) -> TestServer {
let state = AppState::new(pool); let state = AppState::new(pool.clone(), test_runtime(pool));
let app = tc_api::router(state); let app = tc_api::router(state);
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap(); let addr = listener.local_addr().unwrap();
+61
View File
@@ -53,6 +53,67 @@ pub async fn insert(pool: &PgPool, agent: &Agent, policy: &AccessPolicy) -> Resu
Ok(()) Ok(())
} }
pub async fn get(pool: &PgPool, agent_id: AgentId) -> Result<Agent, DbError> {
let row = sqlx::query!(
"SELECT id, workspace_id, name, job_title, system_prompt, avatar,
accent, wallpaper, managed_by, status
FROM agents WHERE id = $1 AND deleted_at IS NULL",
agent_id.as_uuid(),
)
.fetch_one(pool)
.await?;
Ok(Agent {
id: AgentId::from(row.id),
workspace_id: WorkspaceId::from(row.workspace_id),
name: row.name,
job_title: row.job_title,
system_prompt: row.system_prompt,
avatar: row.avatar,
accent: row.accent,
wallpaper: row.wallpaper,
managed_by: UserId::from(row.managed_by),
status: row.status.parse().expect("status CHECK constraint"),
})
}
/// Patch-style profile update (§7.7 Edit profile): only provided fields
/// change; the system prompt is the Job Description textarea verbatim.
#[allow(clippy::too_many_arguments)]
pub async fn update_profile(
pool: &PgPool,
agent_id: AgentId,
name: Option<&str>,
job_title: Option<&str>,
system_prompt: Option<&str>,
avatar: Option<&str>,
accent: Option<&str>,
wallpaper: Option<&str>,
) -> Result<Agent, DbError> {
let result = sqlx::query!(
"UPDATE agents SET
name = COALESCE($2, name),
job_title = COALESCE($3, job_title),
system_prompt = COALESCE($4, system_prompt),
avatar = COALESCE($5, avatar),
accent = COALESCE($6, accent),
wallpaper = COALESCE($7, wallpaper)
WHERE id = $1 AND deleted_at IS NULL",
agent_id.as_uuid(),
name,
job_title,
system_prompt,
avatar,
accent,
wallpaper,
)
.execute(pool)
.await?;
if result.rows_affected() == 0 {
return Err(DbError::NotFound);
}
get(pool, agent_id).await
}
/// The left-rail roster (§4): live agents of a workspace, oldest first. /// The left-rail roster (§4): live agents of a workspace, oldest first.
pub async fn roster(pool: &PgPool, workspace_id: WorkspaceId) -> Result<Vec<Agent>, DbError> { pub async fn roster(pool: &PgPool, workspace_id: WorkspaceId) -> Result<Vec<Agent>, DbError> {
let rows = sqlx::query!( let rows = sqlx::query!(
+116
View File
@@ -0,0 +1,116 @@
use sqlx::PgPool;
use tc_domain::{Message, MessageId, MessageRole, MessageWithSteps, SessionId, Step, StepStatus};
use crate::DbError;
/// Appends a message, assigning the next sequence number atomically.
/// Sessions have a single writer (the run loop) per side, so the rare
/// concurrent collision surfaces as `Conflict` for the caller to retry.
pub async fn append(
pool: &PgPool,
session_id: SessionId,
role: MessageRole,
content: serde_json::Value,
) -> Result<Message, DbError> {
let id = MessageId::new();
let row = sqlx::query!(
"INSERT INTO messages (id, session_id, seq, role, content)
SELECT $1, $2, COALESCE(MAX(seq), 0) + 1, $3, $4
FROM messages WHERE session_id = $2
RETURNING seq, created_at",
id.as_uuid(),
session_id.as_uuid(),
role.as_str(),
content,
)
.fetch_one(pool)
.await?;
Ok(Message {
id,
session_id,
seq: row.seq,
role,
content,
created_at: row.created_at,
})
}
/// Replaces a message's content. The run loop creates the agent reply row
/// before streaming (so steps can attach) and finalizes its text here.
pub async fn set_content(
pool: &PgPool,
message_id: MessageId,
content: serde_json::Value,
) -> Result<(), DbError> {
let result = sqlx::query!(
"UPDATE messages SET content = $2 WHERE id = $1",
message_id.as_uuid(),
content,
)
.execute(pool)
.await?;
if result.rows_affected() == 0 {
return Err(DbError::NotFound);
}
Ok(())
}
/// Full transcript with step traces, oldest first (`?tools=true` shape).
pub async fn history(
pool: &PgPool,
session_id: SessionId,
) -> Result<Vec<MessageWithSteps>, DbError> {
let message_rows = sqlx::query!(
"SELECT id, session_id, seq, role, content, created_at
FROM messages WHERE session_id = $1 ORDER BY seq",
session_id.as_uuid(),
)
.fetch_all(pool)
.await?;
let step_rows = sqlx::query!(
"SELECT s.id, s.message_id, s.seq, s.kind, s.tool_name, s.input,
s.output, s.taint, s.status
FROM steps s
JOIN messages m ON m.id = s.message_id
WHERE m.session_id = $1
ORDER BY m.seq, s.seq",
session_id.as_uuid(),
)
.fetch_all(pool)
.await?;
let mut result: Vec<MessageWithSteps> = message_rows
.into_iter()
.map(|row| MessageWithSteps {
message: Message {
id: MessageId::from(row.id),
session_id: SessionId::from(row.session_id),
seq: row.seq,
role: row.role.parse().expect("role CHECK constraint"),
content: row.content,
created_at: row.created_at,
},
steps: Vec::new(),
})
.collect();
for row in step_rows {
let message_id = MessageId::from(row.message_id);
let step = Step {
id: row.id,
message_id,
seq: row.seq,
kind: row.kind,
tool_name: row.tool_name,
input: row.input,
output: row.output,
taint: row.taint,
status: row.status.parse::<StepStatus>().expect("status values"),
};
if let Some(entry) = result.iter_mut().find(|m| m.message.id == message_id) {
entry.steps.push(step);
}
}
Ok(result)
}
+5
View File
@@ -1,5 +1,10 @@
pub mod agents; pub mod agents;
pub mod audit; pub mod audit;
pub mod credits; pub mod credits;
pub mod messages;
pub mod run_events;
pub mod runs;
pub mod sessions;
pub mod steps;
pub mod users; pub mod users;
pub mod workspaces; pub mod workspaces;
+61
View File
@@ -0,0 +1,61 @@
use sqlx::PgPool;
use uuid::Uuid;
use crate::DbError;
/// One persisted gateway event (§13): the journal the SSE stream and
/// reconnect replay both read from.
#[derive(Debug, Clone, PartialEq)]
pub struct RunEvent {
pub run_id: Uuid,
pub seq: i64,
pub event_type: String,
pub payload: serde_json::Value,
}
/// Persists an event. Called BEFORE the event is emitted to any client so
/// the journal is always at least as complete as what observers saw.
pub async fn append(
pool: &PgPool,
run_id: Uuid,
seq: i64,
event_type: &str,
payload: serde_json::Value,
) -> Result<(), DbError> {
sqlx::query!(
"INSERT INTO run_events (run_id, seq, event_type, payload)
VALUES ($1, $2, $3, $4)",
run_id,
seq,
event_type,
payload,
)
.execute(pool)
.await?;
Ok(())
}
/// Events after a client's `resumeFrom` offset, in order.
pub async fn list_after(
pool: &PgPool,
run_id: Uuid,
after_seq: i64,
) -> Result<Vec<RunEvent>, DbError> {
let rows = sqlx::query!(
"SELECT run_id, seq, event_type, payload
FROM run_events WHERE run_id = $1 AND seq > $2 ORDER BY seq",
run_id,
after_seq,
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|row| RunEvent {
run_id: row.run_id,
seq: row.seq,
event_type: row.event_type,
payload: row.payload,
})
.collect())
}
+89
View File
@@ -0,0 +1,89 @@
use sqlx::PgPool;
use tc_domain::{AgentRun, RunState, SessionId};
use uuid::Uuid;
use crate::DbError;
pub async fn create(pool: &PgPool, session_id: SessionId) -> Result<Uuid, DbError> {
let id = Uuid::now_v7();
sqlx::query!(
"INSERT INTO agent_runs (id, session_id, state) VALUES ($1, $2, 'running')",
id,
session_id.as_uuid(),
)
.execute(pool)
.await?;
Ok(id)
}
pub async fn get(pool: &PgPool, id: Uuid) -> Result<AgentRun, DbError> {
let row = sqlx::query!(
"SELECT id, session_id, state, last_event_id, error FROM agent_runs WHERE id = $1",
id,
)
.fetch_one(pool)
.await?;
Ok(AgentRun {
id: row.id,
session_id: SessionId::from(row.session_id),
state: row.state.parse().expect("state CHECK constraint"),
last_event_id: row.last_event_id,
error: row.error,
})
}
pub async fn set_state(
pool: &PgPool,
id: Uuid,
state: RunState,
error: Option<&str>,
) -> Result<(), DbError> {
let result = sqlx::query!(
"UPDATE agent_runs
SET state = $2, error = COALESCE($3, error), updated_at = now()
WHERE id = $1",
id,
state.as_str(),
error,
)
.execute(pool)
.await?;
if result.rows_affected() == 0 {
return Err(DbError::NotFound);
}
Ok(())
}
/// Records the highest event seq persisted for this run (used by resume).
pub async fn set_last_event(pool: &PgPool, id: Uuid, last_event_id: i64) -> Result<(), DbError> {
sqlx::query!(
"UPDATE agent_runs SET last_event_id = $2, updated_at = now() WHERE id = $1",
id,
last_event_id,
)
.execute(pool)
.await?;
Ok(())
}
/// The most recent run for a session, if any (gateway re-attach).
pub async fn latest_for_session(
pool: &PgPool,
session_id: SessionId,
) -> Result<Option<AgentRun>, DbError> {
let row = sqlx::query!(
"SELECT id, session_id, state, last_event_id, error
FROM agent_runs WHERE session_id = $1
ORDER BY created_at DESC, id DESC LIMIT 1",
session_id.as_uuid(),
)
.fetch_optional(pool)
.await?;
Ok(row.map(|row| AgentRun {
id: row.id,
session_id: SessionId::from(row.session_id),
state: row.state.parse().expect("state CHECK constraint"),
last_event_id: row.last_event_id,
error: row.error,
}))
}
+114
View File
@@ -0,0 +1,114 @@
use sqlx::PgPool;
use tc_domain::{shard_of, AgentId, Session, SessionId, WorkspaceId};
use crate::DbError;
fn row_to_session(
id: uuid::Uuid,
agent_id: uuid::Uuid,
workspace_id: uuid::Uuid,
title: String,
shard: i16,
created_at: time::OffsetDateTime,
last_active_at: time::OffsetDateTime,
) -> Session {
Session {
id: SessionId::from(id),
agent_id: AgentId::from(agent_id),
workspace_id: WorkspaceId::from(workspace_id),
title,
shard: shard as u16,
created_at,
last_active_at,
}
}
pub async fn create(
pool: &PgPool,
agent_id: AgentId,
workspace_id: WorkspaceId,
title: &str,
) -> Result<Session, DbError> {
let id = SessionId::new();
let shard = shard_of(agent_id);
let row = sqlx::query!(
"INSERT INTO sessions (id, agent_id, workspace_id, title, shard)
VALUES ($1, $2, $3, $4, $5)
RETURNING created_at, last_active_at",
id.as_uuid(),
agent_id.as_uuid(),
workspace_id.as_uuid(),
title,
shard as i16,
)
.fetch_one(pool)
.await?;
Ok(Session {
id,
agent_id,
workspace_id,
title: title.to_owned(),
shard,
created_at: row.created_at,
last_active_at: row.last_active_at,
})
}
pub async fn get(pool: &PgPool, id: SessionId) -> Result<Session, DbError> {
let row = sqlx::query!(
"SELECT id, agent_id, workspace_id, title, shard, created_at, last_active_at
FROM sessions WHERE id = $1",
id.as_uuid(),
)
.fetch_one(pool)
.await?;
Ok(row_to_session(
row.id,
row.agent_id,
row.workspace_id,
row.title,
row.shard,
row.created_at,
row.last_active_at,
))
}
/// Sessions column ordering (§6): most recently active first.
pub async fn list_by_agent(pool: &PgPool, agent_id: AgentId) -> Result<Vec<Session>, DbError> {
let rows = sqlx::query!(
"SELECT id, agent_id, workspace_id, title, shard, created_at, last_active_at
FROM sessions WHERE agent_id = $1
ORDER BY last_active_at DESC, id DESC",
agent_id.as_uuid(),
)
.fetch_all(pool)
.await?;
Ok(rows
.into_iter()
.map(|row| {
row_to_session(
row.id,
row.agent_id,
row.workspace_id,
row.title,
row.shard,
row.created_at,
row.last_active_at,
)
})
.collect())
}
/// Marks a session as just-used so it sorts to the top of the column.
pub async fn touch(pool: &PgPool, id: SessionId) -> Result<(), DbError> {
let result = sqlx::query!(
"UPDATE sessions SET last_active_at = clock_timestamp() WHERE id = $1",
id.as_uuid(),
)
.execute(pool)
.await?;
if result.rows_affected() == 0 {
return Err(DbError::NotFound);
}
Ok(())
}
+24
View File
@@ -0,0 +1,24 @@
use sqlx::PgPool;
use tc_domain::Step;
use crate::DbError;
pub async fn append(pool: &PgPool, step: &Step) -> Result<(), DbError> {
sqlx::query!(
"INSERT INTO steps
(id, message_id, seq, kind, tool_name, input, output, taint, status)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)",
step.id,
step.message_id.as_uuid(),
step.seq,
step.kind,
step.tool_name.as_deref(),
step.input.clone(),
step.output.clone(),
&step.taint,
step.status.as_str(),
)
.execute(pool)
.await?;
Ok(())
}
+222
View File
@@ -0,0 +1,222 @@
use serde_json::json;
use tc_db::repo::{agents, messages, run_events, runs, sessions, steps, users, workspaces};
use tc_db::DbError;
use tc_domain::{
AccessPolicy, Agent, AgentId, AgentStatus, MessageRole, Role, RunState, Session, SessionId,
Step, StepStatus, User, UserId, Workspace, WorkspaceId,
};
async fn seeded(pool: &sqlx::PgPool) -> (Workspace, User, Agent) {
let ws = Workspace {
id: WorkspaceId::new(),
name: "Acme".into(),
plan: "team".into(),
};
workspaces::insert(pool, &ws).await.unwrap();
let owner = User {
id: UserId::new(),
workspace_id: ws.id,
email: format!("{}@acme.test", UserId::new()),
role: Role::Owner,
display_name: "Owner".into(),
created_at: time::OffsetDateTime::UNIX_EPOCH,
};
users::insert(pool, &owner).await.unwrap();
let agent = Agent {
id: AgentId::new(),
workspace_id: ws.id,
name: "Scout".into(),
job_title: "Analyst".into(),
system_prompt: "You research.".into(),
avatar: String::new(),
accent: String::new(),
wallpaper: String::new(),
managed_by: owner.id,
status: AgentStatus::Online,
};
agents::insert(pool, &agent, &AccessPolicy::default())
.await
.unwrap();
(ws, owner, agent)
}
async fn new_session(pool: &sqlx::PgPool, agent: &Agent, title: &str) -> Session {
let session = sessions::create(pool, agent.id, agent.workspace_id, title)
.await
.unwrap();
assert_eq!(session.title, title);
session
}
#[tokio::test]
async fn session_create_get_and_shard_are_consistent() {
let pool = tc_testkit::test_pool().await;
let (_, _, agent) = seeded(&pool).await;
let session = new_session(&pool, &agent, "Quarterly research").await;
let fetched = sessions::get(&pool, session.id).await.unwrap();
assert_eq!(fetched, session);
assert_eq!(fetched.shard, tc_domain::shard_of(agent.id));
}
#[tokio::test]
async fn sessions_list_most_recently_active_first() {
let pool = tc_testkit::test_pool().await;
let (_, _, agent) = seeded(&pool).await;
let older = new_session(&pool, &agent, "First").await;
let newer = new_session(&pool, &agent, "Second").await;
let listed = sessions::list_by_agent(&pool, agent.id).await.unwrap();
assert_eq!(
listed.iter().map(|s| s.id).collect::<Vec<_>>(),
vec![newer.id, older.id]
);
// Touching the older session moves it back to the top.
sessions::touch(&pool, older.id).await.unwrap();
let relisted = sessions::list_by_agent(&pool, agent.id).await.unwrap();
assert_eq!(relisted[0].id, older.id);
}
#[tokio::test]
async fn messages_get_increasing_seq_starting_at_one() {
let pool = tc_testkit::test_pool().await;
let (_, _, agent) = seeded(&pool).await;
let session = new_session(&pool, &agent, "Chat").await;
let first = messages::append(&pool, session.id, MessageRole::User, json!({"text": "hi"}))
.await
.unwrap();
let second = messages::append(
&pool,
session.id,
MessageRole::Agent,
json!({"text": "hello!"}),
)
.await
.unwrap();
assert_eq!(first.seq, 1);
assert_eq!(second.seq, 2);
assert_eq!(second.role, MessageRole::Agent);
}
#[tokio::test]
async fn history_returns_messages_with_ordered_steps() {
let pool = tc_testkit::test_pool().await;
let (_, _, agent) = seeded(&pool).await;
let session = new_session(&pool, &agent, "Chat").await;
messages::append(
&pool,
session.id,
MessageRole::User,
json!({"text": "time?"}),
)
.await
.unwrap();
let reply = messages::append(
&pool,
session.id,
MessageRole::Agent,
json!({"text": "It is noon."}),
)
.await
.unwrap();
for (seq, tool) in [(1, "clock.now"), (2, "clock.now")] {
steps::append(
&pool,
&Step {
id: uuid::Uuid::now_v7(),
message_id: reply.id,
seq,
kind: "tool_call".into(),
tool_name: Some(tool.into()),
input: Some(json!({})),
output: Some(json!({"now": "12:00"})),
taint: vec![],
status: StepStatus::Ok,
},
)
.await
.unwrap();
}
let history = messages::history(&pool, session.id).await.unwrap();
assert_eq!(history.len(), 2);
assert!(history[0].steps.is_empty());
let trace = &history[1].steps;
assert_eq!(trace.len(), 2);
assert_eq!(trace[0].seq, 1);
assert_eq!(trace[1].seq, 2);
assert_eq!(trace[0].tool_name.as_deref(), Some("clock.now"));
assert_eq!(trace[0].status, StepStatus::Ok);
}
#[tokio::test]
async fn runs_track_state_transitions() {
let pool = tc_testkit::test_pool().await;
let (_, _, agent) = seeded(&pool).await;
let session = new_session(&pool, &agent, "Chat").await;
let run_id = runs::create(&pool, session.id).await.unwrap();
let run = runs::get(&pool, run_id).await.unwrap();
assert_eq!(run.state, RunState::Running);
runs::set_state(&pool, run_id, RunState::Completed, None)
.await
.unwrap();
let done = runs::get(&pool, run_id).await.unwrap();
assert_eq!(done.state, RunState::Completed);
runs::set_state(&pool, run_id, RunState::Failed, Some("llm unreachable"))
.await
.unwrap();
let failed = runs::get(&pool, run_id).await.unwrap();
assert_eq!(failed.error.as_deref(), Some("llm unreachable"));
}
#[tokio::test]
async fn run_events_replay_from_an_offset() {
let pool = tc_testkit::test_pool().await;
let (_, _, agent) = seeded(&pool).await;
let session = new_session(&pool, &agent, "Chat").await;
let run_id = runs::create(&pool, session.id).await.unwrap();
for (seq, kind) in [(1, "run_started"), (2, "text_delta"), (3, "run_completed")] {
run_events::append(&pool, run_id, seq, kind, json!({"seq": seq}))
.await
.unwrap();
}
let all = run_events::list_after(&pool, run_id, 0).await.unwrap();
assert_eq!(all.len(), 3);
assert_eq!(all[0].event_type, "run_started");
let tail = run_events::list_after(&pool, run_id, 2).await.unwrap();
assert_eq!(tail.len(), 1);
assert_eq!(tail[0].event_type, "run_completed");
assert_eq!(tail[0].seq, 3);
}
#[tokio::test]
async fn duplicate_event_seq_is_a_conflict() {
let pool = tc_testkit::test_pool().await;
let (_, _, agent) = seeded(&pool).await;
let session = new_session(&pool, &agent, "Chat").await;
let run_id = runs::create(&pool, session.id).await.unwrap();
run_events::append(&pool, run_id, 1, "run_started", json!({}))
.await
.unwrap();
let err = run_events::append(&pool, run_id, 1, "run_started", json!({}))
.await
.unwrap_err();
assert!(matches!(err, DbError::Conflict(_)));
}
#[tokio::test]
async fn missing_session_is_not_found() {
let pool = tc_testkit::test_pool().await;
let err = sessions::get(&pool, SessionId::new()).await.unwrap_err();
assert!(matches!(err, DbError::NotFound));
}
+1
View File
@@ -8,6 +8,7 @@ publish.workspace = true
[dependencies] [dependencies]
serde = { workspace = true } serde = { workspace = true }
serde_json = { workspace = true }
thiserror = { workspace = true } thiserror = { workspace = true }
time = { workspace = true } time = { workspace = true }
uuid = { workspace = true } uuid = { workspace = true }
+177
View File
@@ -0,0 +1,177 @@
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use crate::ids::{AgentId, MessageId, SessionId, WorkspaceId};
/// Deterministic hash bucket of an agent id, carried in every SessionKey
/// (§12). Stable across processes and releases: future partitioning relies
/// on it never changing for an existing agent.
pub fn shard_of(agent_id: AgentId) -> u16 {
let bytes = agent_id.as_uuid().into_bytes();
let mut acc: u16 = 0;
for b in bytes {
acc = acc.wrapping_mul(31).wrapping_add(u16::from(b));
}
acc % 16
}
/// A chat session between humans and one agent (spec §14 Session).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Session {
pub id: SessionId,
pub agent_id: AgentId,
pub workspace_id: WorkspaceId,
pub title: String,
pub shard: u16,
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
#[serde(with = "time::serde::rfc3339")]
pub last_active_at: OffsetDateTime,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum MessageRole {
User,
Agent,
System,
}
impl MessageRole {
pub fn as_str(&self) -> &'static str {
match self {
MessageRole::User => "user",
MessageRole::Agent => "agent",
MessageRole::System => "system",
}
}
}
impl std::str::FromStr for MessageRole {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"user" => Ok(MessageRole::User),
"agent" => Ok(MessageRole::Agent),
"system" => Ok(MessageRole::System),
other => Err(format!("unknown message role: {other}")),
}
}
}
/// One transcript message (spec §14 Message). `content` is provider-neutral
/// JSON: `{"text": "..."}` for plain messages.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Message {
pub id: MessageId,
pub session_id: SessionId,
pub seq: i64,
pub role: MessageRole,
pub content: serde_json::Value,
#[serde(with = "time::serde::rfc3339")]
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StepStatus {
Running,
Ok,
Error,
}
impl StepStatus {
pub fn as_str(&self) -> &'static str {
match self {
StepStatus::Running => "running",
StepStatus::Ok => "ok",
StepStatus::Error => "error",
}
}
}
impl std::str::FromStr for StepStatus {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"running" => Ok(StepStatus::Running),
"ok" => Ok(StepStatus::Ok),
"error" => Ok(StepStatus::Error),
other => Err(format!("unknown step status: {other}")),
}
}
}
/// One tool/reasoning step attached to an agent message — the "N steps"
/// trace (§5b). `taint` records untrusted content sources (§15).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Step {
pub id: uuid::Uuid,
pub message_id: MessageId,
pub seq: i32,
pub kind: String,
pub tool_name: Option<String>,
pub input: Option<serde_json::Value>,
pub output: Option<serde_json::Value>,
pub taint: Vec<String>,
pub status: StepStatus,
}
/// A transcript message with its step trace (`?tools=true` history shape).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct MessageWithSteps {
#[serde(flatten)]
pub message: Message,
pub steps: Vec<Step>,
}
/// Lifecycle of one agent run (spec §15 approval interception suspends a
/// run at `AwaitingApproval` until a human decides).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum RunState {
Running,
AwaitingApproval,
Completed,
Failed,
Cancelled,
}
impl RunState {
pub fn as_str(&self) -> &'static str {
match self {
RunState::Running => "running",
RunState::AwaitingApproval => "awaiting_approval",
RunState::Completed => "completed",
RunState::Failed => "failed",
RunState::Cancelled => "cancelled",
}
}
}
impl std::str::FromStr for RunState {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"running" => Ok(RunState::Running),
"awaiting_approval" => Ok(RunState::AwaitingApproval),
"completed" => Ok(RunState::Completed),
"failed" => Ok(RunState::Failed),
"cancelled" => Ok(RunState::Cancelled),
other => Err(format!("unknown run state: {other}")),
}
}
}
/// One execution of the agent loop within a session.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AgentRun {
pub id: uuid::Uuid,
pub session_id: SessionId,
pub state: RunState,
pub last_event_id: i64,
pub error: Option<String>,
}
+4
View File
@@ -5,6 +5,7 @@
//! the fixed set of approval-gated action categories from spec §15. //! the fixed set of approval-gated action categories from spec §15.
mod access; mod access;
mod chat;
mod entities; mod entities;
mod gated; mod gated;
mod ids; mod ids;
@@ -12,6 +13,9 @@ mod role;
mod session_key; mod session_key;
pub use access::{AccessPolicy, AgentScope, HumanScope}; pub use access::{AccessPolicy, AgentScope, HumanScope};
pub use chat::{
shard_of, AgentRun, Message, MessageRole, MessageWithSteps, RunState, Session, Step, StepStatus,
};
pub use entities::{Agent, AgentStatus, User, Workspace}; pub use entities::{Agent, AgentStatus, User, Workspace};
pub use gated::GatedCategory; pub use gated::GatedCategory;
pub use ids::{AgentId, MessageId, SessionId, UserId, WorkspaceId}; pub use ids::{AgentId, MessageId, SessionId, UserId, WorkspaceId};
+28
View File
@@ -0,0 +1,28 @@
[package]
name = "tc-llm"
version = "0.1.0"
edition.workspace = true
rust-version.workspace = true
license.workspace = true
publish.workspace = true
[dependencies]
async-stream = "0.3"
async-trait = "0.1"
eventsource-stream = "0.2"
futures = "0.3"
reqwest = { version = "0.12", default-features = false, features = [
"json",
"rustls-tls",
"stream",
] }
serde = { workspace = true }
serde_json = { workspace = true }
thiserror = { workspace = true }
toml = "0.8"
[dev-dependencies]
tokio = { workspace = true }
[lints]
workspace = true
+175
View File
@@ -0,0 +1,175 @@
//! Anthropic Messages API provider (cloud target).
use async_stream::try_stream;
use eventsource_stream::Eventsource;
use futures::StreamExt;
use serde_json::{json, Value};
use crate::provider::{
ChatRequest, ChatRole, ContentPart, EventStream, LlmError, LlmEvent, LlmProvider, StopReason,
};
pub struct AnthropicProvider {
client: reqwest::Client,
base_url: String,
api_key: String,
}
impl AnthropicProvider {
pub fn new(api_key: String) -> AnthropicProvider {
AnthropicProvider::with_base_url(api_key, "https://api.anthropic.com".into())
}
/// Base URL override exists for self-hosted proxies and tests.
pub fn with_base_url(api_key: String, base_url: String) -> AnthropicProvider {
AnthropicProvider {
client: reqwest::Client::new(),
base_url,
api_key,
}
}
fn wire_messages(request: &ChatRequest) -> Vec<Value> {
request
.messages
.iter()
.map(|message| {
let role = match message.role {
ChatRole::User => "user",
ChatRole::Assistant => "assistant",
};
let content: Vec<Value> = message
.parts
.iter()
.map(|part| match part {
ContentPart::Text(text) => json!({"type": "text", "text": text}),
ContentPart::ToolUse { id, name, input } => {
json!({"type": "tool_use", "id": id, "name": name, "input": input})
}
ContentPart::ToolResult {
tool_use_id,
content,
} => json!({
"type": "tool_result",
"tool_use_id": tool_use_id,
"content": content.to_string(),
}),
})
.collect();
json!({"role": role, "content": content})
})
.collect()
}
}
fn stop_reason(wire: &str) -> StopReason {
match wire {
"tool_use" => StopReason::ToolUse,
"max_tokens" => StopReason::MaxTokens,
_ => StopReason::EndTurn,
}
}
#[async_trait::async_trait]
impl LlmProvider for AnthropicProvider {
async fn stream(&self, request: ChatRequest) -> Result<EventStream, LlmError> {
let tools: Vec<Value> = request
.tools
.iter()
.map(|t| {
json!({
"name": t.name,
"description": t.description,
"input_schema": t.input_schema,
})
})
.collect();
let body = json!({
"model": request.model,
"max_tokens": request.max_tokens,
"system": request.system,
"messages": AnthropicProvider::wire_messages(&request),
"tools": tools,
"stream": true,
});
let response = self
.client
.post(format!("{}/v1/messages", self.base_url))
.header("x-api-key", &self.api_key)
.header("anthropic-version", "2023-06-01")
.json(&body)
.send()
.await
.map_err(|e| LlmError::Transport(e.to_string()))?;
if !response.status().is_success() {
let status = response.status();
let detail = response.text().await.unwrap_or_default();
return Err(LlmError::Api(format!("{status}: {detail}")));
}
let mut sse = response.bytes_stream().eventsource();
let stream = try_stream! {
// tool_use input arrives as accumulated partial JSON between
// content_block_start and content_block_stop.
let mut pending_tool: Option<(String, String, String)> = None; // id, name, json
while let Some(event) = sse.next().await {
let event = event.map_err(|e| LlmError::Transport(e.to_string()))?;
let data: Value = serde_json::from_str(&event.data)
.map_err(|e| LlmError::Wire(format!("{e}: {}", event.data)))?;
match data["type"].as_str().unwrap_or_default() {
"content_block_start" => {
let block = &data["content_block"];
if block["type"] == "tool_use" {
pending_tool = Some((
block["id"].as_str().unwrap_or_default().to_owned(),
block["name"].as_str().unwrap_or_default().to_owned(),
String::new(),
));
}
}
"content_block_delta" => {
let delta = &data["delta"];
match delta["type"].as_str().unwrap_or_default() {
"text_delta" => {
if let Some(text) = delta["text"].as_str() {
yield LlmEvent::TextDelta(text.to_owned());
}
}
"input_json_delta" => {
if let Some((_, _, buf)) = pending_tool.as_mut() {
buf.push_str(delta["partial_json"].as_str().unwrap_or_default());
}
}
_ => {}
}
}
"content_block_stop" => {
if let Some((id, name, buf)) = pending_tool.take() {
let input: Value = if buf.is_empty() {
json!({})
} else {
serde_json::from_str(&buf)
.map_err(|e| LlmError::Wire(format!("tool input: {e}")))?
};
yield LlmEvent::ToolUse { id, name, input };
}
}
"message_delta" => {
if let Some(reason) = data["delta"]["stop_reason"].as_str() {
yield LlmEvent::Stop(stop_reason(reason));
}
}
"error" => {
Err(LlmError::Api(data["error"]["message"]
.as_str()
.unwrap_or("unknown")
.to_owned()))?;
}
_ => {}
}
}
};
Ok(Box::pin(stream))
}
}
+24
View File
@@ -0,0 +1,24 @@
//! LLM access for TeamClaw. One provider trait, three shipping
//! implementations selected by configuration:
//!
//! - `anthropic` — Anthropic Messages API (cloud target)
//! - `openai_compat` — any OpenAI-compatible endpoint (air-gapped vLLM /
//! Ollama / llama.cpp)
//! - `scripted` — deterministic scenario engine; the test/e2e/smoke-test
//! provider that exercises the exact production seam
//!
//! Everything upstream (runtime, gateway, UI) speaks only the
//! provider-neutral types in `provider.rs`.
mod anthropic;
mod openai_compat;
mod provider;
mod scripted;
pub use anthropic::AnthropicProvider;
pub use openai_compat::OpenAiCompatProvider;
pub use provider::{
ChatMessage, ChatRequest, ChatRole, ContentPart, EventStream, LlmError, LlmEvent, LlmProvider,
StopReason, ToolDescriptor,
};
pub use scripted::ScriptedProvider;
+179
View File
@@ -0,0 +1,179 @@
//! OpenAI-compatible chat/completions provider — the air-gapped inference
//! path (vLLM, Ollama, llama.cpp all speak this protocol).
use async_stream::try_stream;
use eventsource_stream::Eventsource;
use futures::StreamExt;
use serde_json::{json, Value};
use crate::provider::{
ChatRequest, ChatRole, ContentPart, EventStream, LlmError, LlmEvent, LlmProvider, StopReason,
};
pub struct OpenAiCompatProvider {
client: reqwest::Client,
base_url: String,
api_key: Option<String>,
}
impl OpenAiCompatProvider {
/// `base_url` includes the version prefix, e.g. `http://local-llm:8000/v1`.
pub fn new(base_url: String, api_key: Option<String>) -> OpenAiCompatProvider {
OpenAiCompatProvider {
client: reqwest::Client::new(),
base_url,
api_key,
}
}
/// Maps provider-neutral messages to the OpenAI wire shape: tool calls
/// ride on assistant messages, tool results become `role: "tool"`.
fn wire_messages(request: &ChatRequest) -> Vec<Value> {
let mut wire = vec![json!({"role": "system", "content": request.system})];
for message in &request.messages {
let mut text = String::new();
let mut tool_calls: Vec<Value> = Vec::new();
for part in &message.parts {
match part {
ContentPart::Text(t) => text.push_str(t),
ContentPart::ToolUse { id, name, input } => tool_calls.push(json!({
"id": id,
"type": "function",
"function": {"name": name, "arguments": input.to_string()},
})),
ContentPart::ToolResult {
tool_use_id,
content,
} => wire.push(json!({
"role": "tool",
"tool_call_id": tool_use_id,
"content": content.to_string(),
})),
}
}
if !text.is_empty() || !tool_calls.is_empty() {
let role = match message.role {
ChatRole::User => "user",
ChatRole::Assistant => "assistant",
};
let mut entry = json!({"role": role, "content": text});
if !tool_calls.is_empty() {
entry["tool_calls"] = Value::Array(tool_calls);
}
wire.push(entry);
}
}
wire
}
}
fn stop_reason(wire: &str) -> StopReason {
match wire {
"tool_calls" => StopReason::ToolUse,
"length" => StopReason::MaxTokens,
_ => StopReason::EndTurn,
}
}
#[async_trait::async_trait]
impl LlmProvider for OpenAiCompatProvider {
async fn stream(&self, request: ChatRequest) -> Result<EventStream, LlmError> {
let tools: Vec<Value> = request
.tools
.iter()
.map(|t| {
json!({
"type": "function",
"function": {
"name": t.name,
"description": t.description,
"parameters": t.input_schema,
},
})
})
.collect();
let mut body = json!({
"model": request.model,
"max_tokens": request.max_tokens,
"messages": OpenAiCompatProvider::wire_messages(&request),
"stream": true,
});
if !tools.is_empty() {
body["tools"] = Value::Array(tools);
}
let mut http = self
.client
.post(format!("{}/chat/completions", self.base_url))
.json(&body);
if let Some(key) = &self.api_key {
http = http.bearer_auth(key);
}
let response = http
.send()
.await
.map_err(|e| LlmError::Transport(e.to_string()))?;
if !response.status().is_success() {
let status = response.status();
let detail = response.text().await.unwrap_or_default();
return Err(LlmError::Api(format!("{status}: {detail}")));
}
let mut sse = response.bytes_stream().eventsource();
let stream = try_stream! {
// Tool-call arguments stream as partial JSON keyed by index.
let mut pending: Vec<(String, String, String)> = Vec::new(); // id, name, args
let mut finish: Option<StopReason> = None;
while let Some(event) = sse.next().await {
let event = event.map_err(|e| LlmError::Transport(e.to_string()))?;
if event.data.trim() == "[DONE]" {
break;
}
let data: Value = serde_json::from_str(&event.data)
.map_err(|e| LlmError::Wire(format!("{e}: {}", event.data)))?;
let choice = &data["choices"][0];
let delta = &choice["delta"];
if let Some(text) = delta["content"].as_str() {
if !text.is_empty() {
yield LlmEvent::TextDelta(text.to_owned());
}
}
if let Some(calls) = delta["tool_calls"].as_array() {
for call in calls {
let index = call["index"].as_u64().unwrap_or(0) as usize;
while pending.len() <= index {
pending.push((String::new(), String::new(), String::new()));
}
let slot = &mut pending[index];
if let Some(id) = call["id"].as_str() {
slot.0 = id.to_owned();
}
if let Some(name) = call["function"]["name"].as_str() {
slot.1.push_str(name);
}
if let Some(args) = call["function"]["arguments"].as_str() {
slot.2.push_str(args);
}
}
}
if let Some(reason) = choice["finish_reason"].as_str() {
finish = Some(stop_reason(reason));
}
}
for (id, name, args) in pending.drain(..) {
if name.is_empty() {
continue;
}
let input: Value = if args.trim().is_empty() {
json!({})
} else {
serde_json::from_str(&args)
.map_err(|e| LlmError::Wire(format!("tool arguments: {e}")))?
};
yield LlmEvent::ToolUse { id, name, input };
}
yield LlmEvent::Stop(finish.unwrap_or(StopReason::EndTurn));
};
Ok(Box::pin(stream))
}
}
+94
View File
@@ -0,0 +1,94 @@
use futures::stream::BoxStream;
use serde::{Deserialize, Serialize};
use serde_json::Value;
/// Provider-neutral conversation role.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum ChatRole {
User,
Assistant,
}
/// Provider-neutral message content. Tool results travel as user-role parts,
/// matching both wire formats' conventions.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum ContentPart {
Text(String),
ToolUse {
id: String,
name: String,
input: Value,
},
ToolResult {
tool_use_id: String,
content: Value,
},
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ChatMessage {
pub role: ChatRole,
pub parts: Vec<ContentPart>,
}
/// A tool offered to the model. `input_schema` is JSON Schema.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ToolDescriptor {
pub name: String,
pub description: String,
pub input_schema: Value,
}
/// The checkpointable, provider-neutral request. This struct is what gets
/// serialized into `agent_runs.checkpoint`, so resume works identically
/// across providers.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ChatRequest {
pub system: String,
pub messages: Vec<ChatMessage>,
pub tools: Vec<ToolDescriptor>,
pub model: String,
pub max_tokens: u32,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StopReason {
EndTurn,
ToolUse,
MaxTokens,
}
/// Streaming events every provider normalizes to.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum LlmEvent {
TextDelta(String),
ToolUse {
id: String,
name: String,
input: Value,
},
Stop(StopReason),
}
#[derive(Debug, thiserror::Error)]
pub enum LlmError {
#[error("scenario error: {0}")]
Scenario(String),
#[error("transport error: {0}")]
Transport(String),
#[error("provider returned an error: {0}")]
Api(String),
#[error("malformed provider response: {0}")]
Wire(String),
}
pub type EventStream = BoxStream<'static, Result<LlmEvent, LlmError>>;
#[async_trait::async_trait]
pub trait LlmProvider: Send + Sync {
async fn stream(&self, request: ChatRequest) -> Result<EventStream, LlmError>;
}
+163
View File
@@ -0,0 +1,163 @@
//! The deterministic scenario provider.
//!
//! A real, config-selectable production provider (`provider = "scripted"`):
//! it powers TDD, Playwright E2E, and the air-gapped smoke-test mode, so
//! tests exercise the exact seam production uses. Scenarios are TOML; a
//! prompt that contains a scenario's `marker` plays that scenario's turns,
//! anything else gets a deterministic echo.
use futures::stream;
use serde::Deserialize;
use serde_json::Value;
use crate::provider::{
ChatRequest, ContentPart, EventStream, LlmError, LlmEvent, LlmProvider, StopReason,
};
#[derive(Debug, Deserialize)]
struct ScenarioFile {
#[serde(default)]
scenario: Vec<Scenario>,
}
#[derive(Debug, Deserialize)]
struct Scenario {
marker: String,
#[serde(default)]
turns: Vec<Turn>,
}
#[derive(Debug, Deserialize)]
struct Turn {
#[serde(default)]
events: Vec<ScriptedEvent>,
}
#[derive(Debug, Clone, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
enum ScriptedEvent {
Text {
text: String,
},
ToolUse {
name: String,
#[serde(default)]
input: Value,
},
}
#[derive(Debug)]
pub struct ScriptedProvider {
scenarios: Vec<Scenario>,
}
impl ScriptedProvider {
pub fn from_toml(toml_source: &str) -> Result<ScriptedProvider, LlmError> {
let file: ScenarioFile =
toml::from_str(toml_source).map_err(|e| LlmError::Scenario(e.to_string()))?;
Ok(ScriptedProvider {
scenarios: file.scenario,
})
}
pub fn from_path(path: &std::path::Path) -> Result<ScriptedProvider, LlmError> {
let source = std::fs::read_to_string(path)
.map_err(|e| LlmError::Scenario(format!("read {}: {e}", path.display())))?;
ScriptedProvider::from_toml(&source)
}
/// Streams a fixed text as word-level deltas to exercise real streaming
/// behavior in every consumer.
fn text_deltas(text: &str, out: &mut Vec<Result<LlmEvent, LlmError>>) {
let mut rest = text;
while !rest.is_empty() {
let cut = rest
.char_indices()
.skip_while(|(_, c)| *c == ' ')
.find(|(_, c)| *c == ' ')
.map(|(i, _)| i)
.unwrap_or(rest.len());
let (chunk, tail) = rest.split_at(cut.max(1));
out.push(Ok(LlmEvent::TextDelta(chunk.to_owned())));
rest = tail;
}
}
}
#[async_trait::async_trait]
impl LlmProvider for ScriptedProvider {
async fn stream(&self, request: ChatRequest) -> Result<EventStream, LlmError> {
let prompt_text: String = request
.messages
.iter()
.flat_map(|m| m.parts.iter())
.filter_map(|p| match p {
ContentPart::Text(t) => Some(t.as_str()),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
// Which leg of a multi-tool conversation is this? One ToolResult in
// the request means turn 0 already played; play turn 1, and so on.
let turn_index = request
.messages
.iter()
.flat_map(|m| m.parts.iter())
.filter(|p| matches!(p, ContentPart::ToolResult { .. }))
.count();
let mut events: Vec<Result<LlmEvent, LlmError>> = Vec::new();
let scenario = self
.scenarios
.iter()
.find(|s| prompt_text.contains(&s.marker));
match scenario {
Some(scenario) => {
match scenario.turns.get(turn_index) {
Some(turn) => {
let mut stopped_for_tool = false;
for (i, event) in turn.events.iter().enumerate() {
match event {
ScriptedEvent::Text { text } => {
Self::text_deltas(text, &mut events);
}
ScriptedEvent::ToolUse { name, input } => {
events.push(Ok(LlmEvent::ToolUse {
id: format!("scripted-tool-{turn_index}-{i}"),
name: name.clone(),
input: input.clone(),
}));
stopped_for_tool = true;
}
}
}
events.push(Ok(LlmEvent::Stop(if stopped_for_tool {
StopReason::ToolUse
} else {
StopReason::EndTurn
})));
}
// More tool round-trips than scripted turns: end cleanly.
None => events.push(Ok(LlmEvent::Stop(StopReason::EndTurn))),
}
}
None => {
let last_user_text = request
.messages
.iter()
.rev()
.flat_map(|m| m.parts.iter())
.find_map(|p| match p {
ContentPart::Text(t) => Some(t.clone()),
_ => None,
})
.unwrap_or_default();
Self::text_deltas(&format!("I received: {last_user_text}"), &mut events);
events.push(Ok(LlmEvent::Stop(StopReason::EndTurn)));
}
}
Ok(Box::pin(stream::iter(events)))
}
}
+71
View File
@@ -0,0 +1,71 @@
//! Opt-in tests against real inference endpoints. Activated with
//! `TC_LIVE_LLM=1` plus `ANTHROPIC_API_KEY` (Anthropic) or
//! `TC_OPENAI_COMPAT_URL` (e.g. a local Ollama at
//! `http://127.0.0.1:11434/v1` with `TC_OPENAI_COMPAT_MODEL` set).
//! CI runs these in a dedicated credentialed job; the default suite uses
//! the scripted provider, which exercises the identical seam.
use futures::StreamExt;
use tc_llm::{
AnthropicProvider, ChatMessage, ChatRequest, ChatRole, ContentPart, LlmEvent, LlmProvider,
OpenAiCompatProvider,
};
fn live_enabled() -> bool {
std::env::var("TC_LIVE_LLM").as_deref() == Ok("1")
}
fn simple_request(model: &str) -> ChatRequest {
ChatRequest {
system: "Answer in exactly one short sentence.".into(),
messages: vec![ChatMessage {
role: ChatRole::User,
parts: vec![ContentPart::Text("Say the word 'pong'.".into())],
}],
tools: vec![],
model: model.into(),
max_tokens: 64,
}
}
async fn collect_text(provider: &dyn LlmProvider, request: ChatRequest) -> String {
let mut stream = provider.stream(request).await.expect("stream opens");
let mut text = String::new();
while let Some(event) = stream.next().await {
if let LlmEvent::TextDelta(t) = event.expect("stream event") {
text.push_str(&t);
}
}
text
}
#[tokio::test]
async fn anthropic_streams_text() {
if !live_enabled() {
eprintln!("skipped: set TC_LIVE_LLM=1 to run");
return;
}
let Ok(key) = std::env::var("ANTHROPIC_API_KEY") else {
eprintln!("skipped: ANTHROPIC_API_KEY not set");
return;
};
let provider = AnthropicProvider::new(key);
let text = collect_text(&provider, simple_request("claude-haiku-4-5-20251001")).await;
assert!(text.to_lowercase().contains("pong"), "got: {text}");
}
#[tokio::test]
async fn openai_compat_streams_text() {
if !live_enabled() {
eprintln!("skipped: set TC_LIVE_LLM=1 to run");
return;
}
let Ok(url) = std::env::var("TC_OPENAI_COMPAT_URL") else {
eprintln!("skipped: TC_OPENAI_COMPAT_URL not set");
return;
};
let model = std::env::var("TC_OPENAI_COMPAT_MODEL").unwrap_or_else(|_| "qwen2.5:0.5b".into());
let provider = OpenAiCompatProvider::new(url, None);
let text = collect_text(&provider, simple_request(&model)).await;
assert!(text.to_lowercase().contains("pong"), "got: {text}");
}
+168
View File
@@ -0,0 +1,168 @@
use futures::StreamExt;
use serde_json::json;
use tc_llm::{
ChatMessage, ChatRequest, ChatRole, ContentPart, LlmError, LlmEvent, LlmProvider,
ScriptedProvider, StopReason,
};
const SCENARIOS: &str = r#"
[[scenario]]
marker = "[[scenario:hello]]"
[[scenario.turns]]
events = [
{ type = "text", text = "Hello! I'm Scout, your research analyst." },
]
[[scenario]]
marker = "[[scenario:tool-time]]"
[[scenario.turns]]
events = [
{ type = "text", text = "Let me check the clock." },
{ type = "tool_use", name = "clock.now", input = {} },
]
[[scenario.turns]]
events = [
{ type = "text", text = "It is exactly noon." },
]
"#;
fn request_with_user_text(text: &str) -> ChatRequest {
ChatRequest {
system: "You are Scout.".into(),
messages: vec![ChatMessage {
role: ChatRole::User,
parts: vec![ContentPart::Text(text.into())],
}],
tools: vec![],
model: "scripted".into(),
max_tokens: 1024,
}
}
async fn collect(provider: &ScriptedProvider, request: ChatRequest) -> Vec<LlmEvent> {
let mut stream = provider.stream(request).await.unwrap();
let mut events = Vec::new();
while let Some(event) = stream.next().await {
events.push(event.unwrap());
}
events
}
fn joined_text(events: &[LlmEvent]) -> String {
events
.iter()
.filter_map(|e| match e {
LlmEvent::TextDelta(t) => Some(t.as_str()),
_ => None,
})
.collect()
}
#[tokio::test]
async fn unmatched_prompts_get_a_deterministic_echo() {
let provider = ScriptedProvider::from_toml(SCENARIOS).unwrap();
let events = collect(&provider, request_with_user_text("ping")).await;
assert_eq!(joined_text(&events), "I received: ping");
assert!(matches!(
events.last(),
Some(LlmEvent::Stop(StopReason::EndTurn))
));
}
#[tokio::test]
async fn text_is_streamed_as_multiple_deltas() {
let provider = ScriptedProvider::from_toml(SCENARIOS).unwrap();
let events = collect(&provider, request_with_user_text("hi [[scenario:hello]]")).await;
let delta_count = events
.iter()
.filter(|e| matches!(e, LlmEvent::TextDelta(_)))
.count();
assert!(
delta_count > 1,
"expected word-level streaming, got {delta_count}"
);
assert_eq!(
joined_text(&events),
"Hello! I'm Scout, your research analyst."
);
}
#[tokio::test]
async fn tool_scenario_emits_tool_use_then_continues_after_result() {
let provider = ScriptedProvider::from_toml(SCENARIOS).unwrap();
// First leg: the scripted model asks for the clock tool.
let first = collect(
&provider,
request_with_user_text("what time is it? [[scenario:tool-time]]"),
)
.await;
assert_eq!(joined_text(&first), "Let me check the clock.");
let tool_use = first
.iter()
.find_map(|e| match e {
LlmEvent::ToolUse { id, name, .. } => Some((id.clone(), name.clone())),
_ => None,
})
.expect("a tool_use event");
assert_eq!(tool_use.1, "clock.now");
assert!(matches!(
first.last(),
Some(LlmEvent::Stop(StopReason::ToolUse))
));
// Second leg: request now carries the tool result; the scripted model
// continues with the next turn.
let mut request = request_with_user_text("what time is it? [[scenario:tool-time]]");
request.messages.push(ChatMessage {
role: ChatRole::Assistant,
parts: vec![ContentPart::ToolUse {
id: tool_use.0.clone(),
name: tool_use.1.clone(),
input: json!({}),
}],
});
request.messages.push(ChatMessage {
role: ChatRole::User,
parts: vec![ContentPart::ToolResult {
tool_use_id: tool_use.0,
content: json!({"now": "12:00"}),
}],
});
let second = collect(&provider, request).await;
assert_eq!(joined_text(&second), "It is exactly noon.");
assert!(matches!(
second.last(),
Some(LlmEvent::Stop(StopReason::EndTurn))
));
}
#[tokio::test]
async fn exhausted_turns_fall_back_to_end_turn() {
let provider = ScriptedProvider::from_toml(SCENARIOS).unwrap();
let mut request = request_with_user_text("x [[scenario:hello]]");
// Pretend two tool round-trips already happened: only one turn exists.
for _ in 0..2 {
request.messages.push(ChatMessage {
role: ChatRole::User,
parts: vec![ContentPart::ToolResult {
tool_use_id: "t1".into(),
content: json!({}),
}],
});
}
let events = collect(&provider, request).await;
assert!(matches!(
events.last(),
Some(LlmEvent::Stop(StopReason::EndTurn))
));
}
#[test]
fn invalid_scenario_toml_is_a_clear_error() {
let err = ScriptedProvider::from_toml("not [valid toml").unwrap_err();
assert!(matches!(err, LlmError::Scenario(_)));
}
+27
View File
@@ -0,0 +1,27 @@
[package]
name = "tc-runtime"
version = "0.1.0"
edition.workspace = true
rust-version.workspace = true
license.workspace = true
publish.workspace = true
[dependencies]
async-trait = "0.1"
futures = "0.3"
serde = { workspace = true }
serde_json = { workspace = true }
sqlx = { workspace = true }
tc-db = { path = "../tc-db" }
tc-domain = { path = "../tc-domain" }
tc-llm = { path = "../tc-llm" }
thiserror = { workspace = true }
time = { workspace = true }
tokio = { workspace = true }
uuid = { workspace = true }
[dev-dependencies]
tc-testkit = { path = "../tc-testkit" }
[lints]
workspace = true
+53
View File
@@ -0,0 +1,53 @@
use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;
/// One gateway event with its journal sequence number. The wire contract
/// with the frontend (`frontend/src/lib/gateway/events.ts` mirrors it).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunEventEnvelope {
pub seq: i64,
#[serde(flatten)]
pub event: RunEventBody,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum RunEventBody {
RunStarted {
run_id: Uuid,
},
TextDelta {
delta: String,
},
StepStarted {
step_seq: i32,
tool: String,
input: Value,
},
StepFinished {
step_seq: i32,
status: String,
output: Value,
},
RunCompleted {
message_id: String,
},
Error {
message: String,
},
}
impl RunEventBody {
/// Stable name stored in `run_events.event_type`.
pub fn type_name(&self) -> &'static str {
match self {
RunEventBody::RunStarted { .. } => "run_started",
RunEventBody::TextDelta { .. } => "text_delta",
RunEventBody::StepStarted { .. } => "step_started",
RunEventBody::StepFinished { .. } => "step_finished",
RunEventBody::RunCompleted { .. } => "run_completed",
RunEventBody::Error { .. } => "error",
}
}
}
+11
View File
@@ -0,0 +1,11 @@
//! The agent run loop: takes a user message, drives the LLM provider,
//! executes tools, and journals every event before any observer sees it
//! (the gateway streams exactly this journal, live or replayed).
mod events;
mod runtime;
mod tools;
pub use events::{RunEventBody, RunEventEnvelope};
pub use runtime::{Runtime, RuntimeConfig, RuntimeError, StartedRun};
pub use tools::{ClockNow, Tool, ToolRegistry};
+347
View File
@@ -0,0 +1,347 @@
use std::collections::HashMap;
use std::sync::Arc;
use futures::StreamExt;
use serde_json::{json, Value};
use sqlx::PgPool;
use tc_db::repo::{messages, run_events, runs, sessions, steps};
use tc_db::DbError;
use tc_domain::{MessageRole, MessageWithSteps, SessionId, Step, StepStatus};
use tc_llm::{ChatMessage, ChatRequest, ChatRole, ContentPart, LlmEvent, LlmProvider, StopReason};
use tokio::sync::{broadcast, Mutex};
use uuid::Uuid;
use crate::events::{RunEventBody, RunEventEnvelope};
use crate::tools::ToolRegistry;
#[derive(Debug, Clone)]
pub struct RuntimeConfig {
pub model: String,
pub max_tokens: u32,
}
#[derive(Debug, thiserror::Error)]
pub enum RuntimeError {
#[error(transparent)]
Db(#[from] DbError),
#[error("llm error: {0}")]
Llm(String),
}
/// Handle returned to the caller: the run id plus a live event receiver
/// subscribed before the loop starts (no events can be missed).
pub struct StartedRun {
pub run_id: Uuid,
pub events: broadcast::Receiver<RunEventEnvelope>,
}
/// Owns active run channels; one instance lives in the server state.
/// Cheap to clone (shared inner state).
#[derive(Clone)]
pub struct Runtime {
inner: Arc<RuntimeInner>,
}
struct RuntimeInner {
pool: PgPool,
provider: Arc<dyn LlmProvider>,
tools: Arc<ToolRegistry>,
config: RuntimeConfig,
channels: Mutex<HashMap<Uuid, broadcast::Sender<RunEventEnvelope>>>,
}
impl Runtime {
pub fn new(pool: PgPool, provider: Arc<dyn LlmProvider>, config: RuntimeConfig) -> Runtime {
Runtime {
inner: Arc::new(RuntimeInner {
pool,
provider,
tools: Arc::new(ToolRegistry::default()),
config,
channels: Mutex::new(HashMap::new()),
}),
}
}
/// Live-attach to a run that is still streaming.
pub async fn subscribe(&self, run_id: Uuid) -> Option<broadcast::Receiver<RunEventEnvelope>> {
self.inner
.channels
.lock()
.await
.get(&run_id)
.map(|s| s.subscribe())
}
/// Starts a run for a user message and spawns the loop.
pub async fn send_message(
&self,
session_id: SessionId,
user_text: &str,
) -> Result<StartedRun, RuntimeError> {
let inner = &self.inner;
let session = sessions::get(&inner.pool, session_id).await?;
let agent = tc_db::repo::agents::get(&inner.pool, session.agent_id).await?;
let history = messages::history(&inner.pool, session_id).await?;
let run_id = runs::create(&inner.pool, session_id).await?;
let (sender, receiver) = broadcast::channel(1024);
inner.channels.lock().await.insert(run_id, sender.clone());
let runtime = self.clone();
let user_text = user_text.to_owned();
tokio::spawn(async move {
let result = runtime
.run_loop(
run_id,
session_id,
&agent.system_prompt,
history,
&user_text,
&sender,
)
.await;
if let Err(error) = result {
let _ = runs::set_state(
&runtime.inner.pool,
run_id,
tc_domain::RunState::Failed,
Some(&error.to_string()),
)
.await;
let mut seq = last_journaled_seq(&runtime.inner.pool, run_id).await;
let _ = runtime
.emit(
run_id,
&sender,
&mut seq,
RunEventBody::Error {
message: error.to_string(),
},
)
.await;
}
runtime.inner.channels.lock().await.remove(&run_id);
});
Ok(StartedRun {
run_id,
events: receiver,
})
}
/// Journals an event, then broadcasts it. Persist-before-emit is the
/// invariant that makes replay equal to what live observers saw.
async fn emit(
&self,
run_id: Uuid,
sender: &broadcast::Sender<RunEventEnvelope>,
seq: &mut i64,
event: RunEventBody,
) -> Result<(), RuntimeError> {
*seq += 1;
let payload = serde_json::to_value(&event).expect("event serializes");
run_events::append(&self.inner.pool, run_id, *seq, event.type_name(), payload).await?;
runs::set_last_event(&self.inner.pool, run_id, *seq).await?;
let _ = sender.send(RunEventEnvelope { seq: *seq, event });
Ok(())
}
async fn run_loop(
&self,
run_id: Uuid,
session_id: SessionId,
system_prompt: &str,
history: Vec<MessageWithSteps>,
user_text: &str,
sender: &broadcast::Sender<RunEventEnvelope>,
) -> Result<(), RuntimeError> {
let mut seq: i64 = 0;
self.emit(
run_id,
sender,
&mut seq,
RunEventBody::RunStarted { run_id },
)
.await?;
messages::append(
&self.inner.pool,
session_id,
MessageRole::User,
json!({"text": user_text}),
)
.await?;
sessions::touch(&self.inner.pool, session_id).await?;
// The reply row exists up front so steps can attach to it while the
// run streams; its text is finalized at the end.
let reply = messages::append(
&self.inner.pool,
session_id,
MessageRole::Agent,
json!({"text": ""}),
)
.await?;
let mut request = ChatRequest {
system: system_prompt.to_owned(),
messages: chat_messages(&history, user_text),
tools: self.inner.tools.descriptors(),
model: self.inner.config.model.clone(),
max_tokens: self.inner.config.max_tokens,
};
let mut full_text = String::new();
let mut step_seq: i32 = 0;
loop {
let mut stream = self
.inner
.provider
.stream(request.clone())
.await
.map_err(|e| RuntimeError::Llm(e.to_string()))?;
let mut tool_uses: Vec<(String, String, Value)> = Vec::new();
let mut stop = StopReason::EndTurn;
while let Some(event) = stream.next().await {
match event.map_err(|e| RuntimeError::Llm(e.to_string()))? {
LlmEvent::TextDelta(delta) => {
full_text.push_str(&delta);
self.emit(run_id, sender, &mut seq, RunEventBody::TextDelta { delta })
.await?;
}
LlmEvent::ToolUse { id, name, input } => {
tool_uses.push((id, name, input));
}
LlmEvent::Stop(reason) => stop = reason,
}
}
if tool_uses.is_empty() || stop != StopReason::ToolUse {
break;
}
let mut assistant_parts = Vec::new();
let mut result_parts = Vec::new();
for (tool_use_id, name, input) in tool_uses {
step_seq += 1;
self.emit(
run_id,
sender,
&mut seq,
RunEventBody::StepStarted {
step_seq,
tool: name.clone(),
input: input.clone(),
},
)
.await?;
let outcome = self.inner.tools.execute(&name, input.clone()).await;
let (status, output) = match outcome {
Ok(value) => (StepStatus::Ok, value),
Err(message) => (StepStatus::Error, json!({"error": message})),
};
steps::append(
&self.inner.pool,
&Step {
id: Uuid::now_v7(),
message_id: reply.id,
seq: step_seq,
kind: "tool_call".into(),
tool_name: Some(name.clone()),
input: Some(input.clone()),
output: Some(output.clone()),
taint: vec![],
status,
},
)
.await?;
self.emit(
run_id,
sender,
&mut seq,
RunEventBody::StepFinished {
step_seq,
status: status.as_str().to_owned(),
output: output.clone(),
},
)
.await?;
assistant_parts.push(ContentPart::ToolUse {
id: tool_use_id.clone(),
name,
input,
});
result_parts.push(ContentPart::ToolResult {
tool_use_id,
content: output,
});
}
request.messages.push(ChatMessage {
role: ChatRole::Assistant,
parts: assistant_parts,
});
request.messages.push(ChatMessage {
role: ChatRole::User,
parts: result_parts,
});
}
messages::set_content(&self.inner.pool, reply.id, json!({"text": full_text})).await?;
runs::set_state(
&self.inner.pool,
run_id,
tc_domain::RunState::Completed,
None,
)
.await?;
self.emit(
run_id,
sender,
&mut seq,
RunEventBody::RunCompleted {
message_id: reply.id.to_string(),
},
)
.await?;
Ok(())
}
}
/// Maps persisted history plus the new user message into provider-neutral
/// chat messages. Past step traces are display data; the model sees text.
fn chat_messages(history: &[MessageWithSteps], user_text: &str) -> Vec<ChatMessage> {
let mut out: Vec<ChatMessage> = history
.iter()
.filter_map(|entry| {
let text = entry.message.content["text"].as_str().unwrap_or_default();
if text.is_empty() {
return None;
}
let role = match entry.message.role {
MessageRole::User | MessageRole::System => ChatRole::User,
MessageRole::Agent => ChatRole::Assistant,
};
Some(ChatMessage {
role,
parts: vec![ContentPart::Text(text.to_owned())],
})
})
.collect();
out.push(ChatMessage {
role: ChatRole::User,
parts: vec![ContentPart::Text(user_text.to_owned())],
});
out
}
/// Sequence continuation for the failure path: errors are journaled after
/// whatever the loop already wrote.
async fn last_journaled_seq(pool: &PgPool, run_id: Uuid) -> i64 {
runs::get(pool, run_id)
.await
.map(|r| r.last_event_id)
.unwrap_or(0)
}
+69
View File
@@ -0,0 +1,69 @@
//! Built-in ungated tools. P1 ships server-side tools with no external
//! effects; sandboxed environment tools and the gate policy arrive in P2.
use std::collections::HashMap;
use std::sync::Arc;
use serde_json::{json, Value};
use tc_llm::ToolDescriptor;
#[async_trait::async_trait]
pub trait Tool: Send + Sync {
fn descriptor(&self) -> ToolDescriptor;
async fn execute(&self, input: Value) -> Result<Value, String>;
}
/// Current UTC time — the simplest real tool; lets scenarios and the UI
/// exercise full step traces without a sandbox.
pub struct ClockNow;
#[async_trait::async_trait]
impl Tool for ClockNow {
fn descriptor(&self) -> ToolDescriptor {
ToolDescriptor {
name: "clock.now".into(),
description: "Returns the current UTC date and time.".into(),
input_schema: json!({"type": "object", "properties": {}}),
}
}
async fn execute(&self, _input: Value) -> Result<Value, String> {
let now = time::OffsetDateTime::now_utc()
.format(&time::format_description::well_known::Rfc3339)
.map_err(|e| e.to_string())?;
Ok(json!({ "now": now }))
}
}
pub struct ToolRegistry {
tools: HashMap<String, Arc<dyn Tool>>,
}
impl Default for ToolRegistry {
fn default() -> Self {
let mut registry = ToolRegistry {
tools: HashMap::new(),
};
registry.register(Arc::new(ClockNow));
registry
}
}
impl ToolRegistry {
pub fn register(&mut self, tool: Arc<dyn Tool>) {
self.tools.insert(tool.descriptor().name, tool);
}
pub fn descriptors(&self) -> Vec<ToolDescriptor> {
let mut all: Vec<ToolDescriptor> = self.tools.values().map(|t| t.descriptor()).collect();
all.sort_by(|a, b| a.name.cmp(&b.name));
all
}
pub async fn execute(&self, name: &str, input: Value) -> Result<Value, String> {
match self.tools.get(name) {
Some(tool) => tool.execute(input).await,
None => Err(format!("unknown tool: {name}")),
}
}
}
+237
View File
@@ -0,0 +1,237 @@
use std::sync::Arc;
use tc_domain::{
AccessPolicy, Agent, AgentId, AgentStatus, MessageRole, Role, RunState, User, UserId,
Workspace, WorkspaceId,
};
use tc_llm::ScriptedProvider;
use tc_runtime::{RunEventBody, Runtime, RuntimeConfig};
const SCENARIOS: &str = r#"
[[scenario]]
marker = "[[scenario:tool-time]]"
[[scenario.turns]]
events = [
{ type = "text", text = "Let me check the clock." },
{ type = "tool_use", name = "clock.now", input = {} },
]
[[scenario.turns]]
events = [
{ type = "text", text = " It is now known." },
]
"#;
async fn seeded(pool: &sqlx::PgPool) -> Agent {
let ws = Workspace {
id: WorkspaceId::new(),
name: "Acme".into(),
plan: "team".into(),
};
tc_db::repo::workspaces::insert(pool, &ws).await.unwrap();
let owner = User {
id: UserId::new(),
workspace_id: ws.id,
email: format!("{}@acme.test", UserId::new()),
role: Role::Owner,
display_name: "Owner".into(),
created_at: time::OffsetDateTime::UNIX_EPOCH,
};
tc_db::repo::users::insert(pool, &owner).await.unwrap();
let agent = Agent {
id: AgentId::new(),
workspace_id: ws.id,
name: "Scout".into(),
job_title: "Analyst".into(),
system_prompt: "You are concise.".into(),
avatar: String::new(),
accent: String::new(),
wallpaper: String::new(),
managed_by: owner.id,
status: AgentStatus::Online,
};
tc_db::repo::agents::insert(pool, &agent, &AccessPolicy::default())
.await
.unwrap();
agent
}
fn runtime(pool: sqlx::PgPool) -> Runtime {
Runtime::new(
pool,
Arc::new(ScriptedProvider::from_toml(SCENARIOS).unwrap()),
RuntimeConfig {
model: "scripted".into(),
max_tokens: 1024,
},
)
}
async fn drain(
mut rx: tokio::sync::broadcast::Receiver<tc_runtime::RunEventEnvelope>,
) -> Vec<tc_runtime::RunEventEnvelope> {
let mut events = Vec::new();
while let Ok(envelope) = rx.recv().await {
let done = matches!(
envelope.event,
RunEventBody::RunCompleted { .. } | RunEventBody::Error { .. }
);
events.push(envelope);
if done {
break;
}
}
events
}
#[tokio::test]
async fn plain_message_streams_and_persists_a_reply() {
let pool = tc_testkit::test_pool().await;
let agent = seeded(&pool).await;
let session = tc_db::repo::sessions::create(&pool, agent.id, agent.workspace_id, "Chat")
.await
.unwrap();
let rt = runtime(pool.clone());
let started = rt.send_message(session.id, "hello there").await.unwrap();
let events = drain(started.events).await;
// Envelope sequence is strictly increasing from 1.
let seqs: Vec<i64> = events.iter().map(|e| e.seq).collect();
assert_eq!(seqs, (1..=seqs.len() as i64).collect::<Vec<_>>());
assert!(matches!(events[0].event, RunEventBody::RunStarted { .. }));
let text: String = events
.iter()
.filter_map(|e| match &e.event {
RunEventBody::TextDelta { delta } => Some(delta.as_str()),
_ => None,
})
.collect();
assert_eq!(text, "I received: hello there");
// Transcript persisted: user message + agent reply with that text.
let history = tc_db::repo::messages::history(&pool, session.id)
.await
.unwrap();
assert_eq!(history.len(), 2);
assert_eq!(history[0].message.role, MessageRole::User);
assert_eq!(history[1].message.role, MessageRole::Agent);
assert_eq!(
history[1].message.content["text"],
"I received: hello there"
);
let run = tc_db::repo::runs::get(&pool, started.run_id).await.unwrap();
assert_eq!(run.state, RunState::Completed);
}
#[tokio::test]
async fn tool_scenario_executes_real_tool_and_records_steps() {
let pool = tc_testkit::test_pool().await;
let agent = seeded(&pool).await;
let session = tc_db::repo::sessions::create(&pool, agent.id, agent.workspace_id, "Chat")
.await
.unwrap();
let rt = runtime(pool.clone());
let started = rt
.send_message(session.id, "time? [[scenario:tool-time]]")
.await
.unwrap();
let events = drain(started.events).await;
let step_started = events.iter().find_map(|e| match &e.event {
RunEventBody::StepStarted { tool, .. } => Some(tool.clone()),
_ => None,
});
assert_eq!(step_started.as_deref(), Some("clock.now"));
let step_finished = events.iter().find_map(|e| match &e.event {
RunEventBody::StepFinished { status, output, .. } => Some((status.clone(), output.clone())),
_ => None,
});
let (status, output) = step_finished.expect("step finished event");
assert_eq!(status, "ok");
assert!(output["now"].is_string(), "clock output: {output}");
// Both text legs land in one agent message.
let history = tc_db::repo::messages::history(&pool, session.id)
.await
.unwrap();
let reply = &history[1];
let text = reply.message.content["text"].as_str().unwrap();
assert!(text.contains("Let me check the clock."), "got: {text}");
assert!(text.contains("It is now known."), "got: {text}");
// The step trace is persisted on the reply (the "N steps" UI source).
assert_eq!(reply.steps.len(), 1);
assert_eq!(reply.steps[0].tool_name.as_deref(), Some("clock.now"));
assert_eq!(reply.steps[0].status, tc_domain::StepStatus::Ok);
}
#[tokio::test]
async fn journal_matches_what_subscribers_saw() {
let pool = tc_testkit::test_pool().await;
let agent = seeded(&pool).await;
let session = tc_db::repo::sessions::create(&pool, agent.id, agent.workspace_id, "Chat")
.await
.unwrap();
let rt = runtime(pool.clone());
let started = rt.send_message(session.id, "ping").await.unwrap();
let live = drain(started.events).await;
// Replay from the journal reproduces the exact live sequence (the
// property reconnect/resumeFrom depends on).
let journal = tc_db::repo::run_events::list_after(&pool, started.run_id, 0)
.await
.unwrap();
assert_eq!(journal.len(), live.len());
for (persisted, observed) in journal.iter().zip(live.iter()) {
assert_eq!(persisted.seq, observed.seq);
let live_json = serde_json::to_value(&observed.event).unwrap();
assert_eq!(persisted.event_type, live_json["type"].as_str().unwrap());
}
}
#[tokio::test]
async fn unknown_tool_fails_the_step_but_not_the_run() {
let pool = tc_testkit::test_pool().await;
let agent = seeded(&pool).await;
let session = tc_db::repo::sessions::create(&pool, agent.id, agent.workspace_id, "Chat")
.await
.unwrap();
let scenarios = r#"
[[scenario]]
marker = "[[scenario:bad-tool]]"
[[scenario.turns]]
events = [ { type = "tool_use", name = "no.such.tool", input = {} } ]
[[scenario.turns]]
events = [ { type = "text", text = "I could not use that tool." } ]
"#;
let rt = Runtime::new(
pool.clone(),
Arc::new(ScriptedProvider::from_toml(scenarios).unwrap()),
RuntimeConfig {
model: "scripted".into(),
max_tokens: 1024,
},
);
let started = rt
.send_message(session.id, "x [[scenario:bad-tool]]")
.await
.unwrap();
let events = drain(started.events).await;
let failed_step = events.iter().any(
|e| matches!(&e.event, RunEventBody::StepFinished { status, .. } if status == "error"),
);
assert!(failed_step, "expected an errored step");
// The error went back to the model as a tool result and it continued.
let run = tc_db::repo::runs::get(&pool, started.run_id).await.unwrap();
assert_eq!(run.state, RunState::Completed);
}
+18 -2
View File
@@ -1,9 +1,25 @@
# Scenario script for the deterministic `scripted` LLM provider. # Scenario script for the deterministic `scripted` LLM provider.
# Consumed by tc-llm's ScriptedProvider from P1 onward; each entry maps a # A prompt containing a `marker` plays that scenario's turns; each tool
# prompt marker to the exact event stream the provider emits. # round-trip advances to the next turn. Unmatched prompts get an echo.
[[scenario]] [[scenario]]
marker = "[[scenario:hello]]" marker = "[[scenario:hello]]"
[[scenario.turns]]
events = [ events = [
{ type = "text", text = "Hello! I'm Scout, your research analyst." }, { type = "text", text = "Hello! I'm Scout, your research analyst." },
] ]
[[scenario]]
marker = "[[scenario:tool-time]]"
[[scenario.turns]]
events = [
{ type = "text", text = "Let me check the clock." },
{ type = "tool_use", name = "clock.now", input = {} },
]
[[scenario.turns]]
events = [
{ type = "text", text = "Done — I checked the current time for you." },
]
+13
View File
@@ -0,0 +1,13 @@
-- Gateway event journal (§13 POST /gateway). Every stream event is persisted
-- here BEFORE it is emitted, so live streaming and reconnect replay
-- (`resumeFrom`) read the same data and the gateway stays the single audited
-- channel (§15).
CREATE TABLE run_events (
run_id UUID NOT NULL REFERENCES agent_runs (id),
seq BIGINT NOT NULL,
event_type TEXT NOT NULL,
payload JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
PRIMARY KEY (run_id, seq)
);