dguiducci/Skald-Circle
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: dguiducci/Skald-Circle:088c0c01cb2fc56b9b31c712c4868b710a7c212f
Branches Tags
dguiducci/Skald-Circle:main dguiducci/Skald-Circle:release
...
compare: dguiducci/Skald-Circle:587958ffe0c9281bfad6ab723884cb4728b99da1
Branches Tags
dguiducci/Skald-Circle:main dguiducci/Skald-Circle:release
3 Commits
088c0c01cb ... 587958ffe0
Author SHA1 Message Date
dguiducci 587958ffe0 llm request tracking, user context cleanup, minor fixes 2026-07-11 00:23:14 +01:00
dguiducci 5936cf0b2e multiuser part 2 2026-07-10 22:15:35 +01:00
dguiducci c72f18b0d6 Multuser part 1 2026-07-10 22:05:25 +01:00
40 changed files with 939 additions and 969 deletions
Expand all files Collapse all files
Cargo.lock
Generated
+14 -487
View File
@@ -88,29 +88,6 @@ version = "1.0.102"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
[[package]]
name = "aquamarine"
version = "0.6.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0f50776554130342de4836ba542aa85a4ddb361690d7e8df13774d7284c3d5c2"
dependencies = [
"include_dir",
"itertools 0.10.5",
"proc-macro-error2",
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "ar_archive_writer"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7eb93bbb63b9c227414f6eb3a0adfddca591a8ce1e9b60661bb08969b87e340b"
dependencies = [
"object",
]
[[package]]
name = "argon2"
version = "0.5.3"
@@ -432,12 +409,6 @@ version = "1.5.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
[[package]]
name = "byteorder-lite"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495"
[[package]]
name = "bytes"
version = "1.11.1"
@@ -668,15 +639,6 @@ version = "0.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0c9ea0ac24bc397ab3c98583a3c9ba74fa56b09a4449bbe172b9b1ddb016027a"
[[package]]
name = "colored"
version = "3.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "combine"
version = "4.6.7"
@@ -1147,34 +1109,13 @@ dependencies = [
"serde_core",
]
[[package]]
name = "derive_more"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05"
dependencies = [
"derive_more-impl 1.0.0",
]
[[package]]
name = "derive_more"
version = "2.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d751e9e49156b02b44f9c1815bcb94b984cdcc4396ecc32521c739452808b134"
dependencies = [
"derive_more-impl 2.1.1",
]
[[package]]
name = "derive_more-impl"
version = "1.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.117",
"unicode-xid",
"derive_more-impl",
]
[[package]]
@@ -1335,16 +1276,6 @@ dependencies = [
"serde",
]
[[package]]
name = "dptree"
version = "0.5.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "db96968fcf52fe063a98c75df1d1f2b1fba304e7ae29b72fdc81c1165b7e2fd0"
dependencies = [
"colored",
"futures",
]
[[package]]
name = "dtoa"
version = "1.0.11"
@@ -1475,16 +1406,6 @@ version = "1.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
[[package]]
name = "erasable"
version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "437cfb75878119ed8265685c41a115724eae43fb7cc5a0bf0e4ecc3b803af1c4"
dependencies = [
"autocfg",
"scopeguard",
]
[[package]]
name = "erased-serde"
version = "0.4.10"
@@ -2267,7 +2188,7 @@ checksum = "ec9d92d097f4749b64e8cc33d924d9f40a2d4eb91402b458014b781f5733d60f"
name = "honcho-client"
version = "0.1.0"
dependencies = [
"reqwest 0.13.4",
"reqwest",
"serde",
"serde_json",
"tracing",
@@ -2390,7 +2311,6 @@ dependencies = [
"tokio",
"tokio-rustls",
"tower-service",
"webpki-roots 1.0.7",
]
[[package]]
@@ -2567,38 +2487,6 @@ dependencies = [
"icu_properties",
]
[[package]]
name = "image"
version = "0.25.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "85ab80394333c02fe689eaf900ab500fbd0c2213da414687ebf995a65d5a6104"
dependencies = [
"bytemuck",
"byteorder-lite",
"moxcms",
"num-traits",
"png 0.18.1",
]
[[package]]
name = "include_dir"
version = "0.7.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "923d117408f1e49d914f1a379a309cffe4f18c05cf4e3d12e613a15fc81bd0dd"
dependencies = [
"include_dir_macros",
]
[[package]]
name = "include_dir_macros"
version = "0.7.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7cab85a7ed0bd5f0e76d93846e0147172bed2e2d3f859bcc33a8d9699cad1a75"
dependencies = [
"proc-macro2",
"quote",
]
[[package]]
name = "indexmap"
version = "1.9.3"
@@ -3031,7 +2919,7 @@ dependencies = [
"anyhow",
"async-trait",
"core-api",
"reqwest 0.13.4",
"reqwest",
"serde",
"serde_json",
"tracing",
@@ -3052,12 +2940,6 @@ version = "0.4.29"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897"
[[package]]
name = "lru-slab"
version = "0.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154"
[[package]]
name = "managed"
version = "0.8.0"
@@ -3097,7 +2979,7 @@ dependencies = [
"anyhow",
"async-trait",
"base64 0.22.1",
"reqwest 0.13.4",
"reqwest",
"serde",
"serde_json",
"tokio",
@@ -3179,16 +3061,6 @@ dependencies = [
"windows-sys 0.61.2",
]
[[package]]
name = "moxcms"
version = "0.8.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "bb85c154ba489f01b25c0d36ae69a87e4a1c73a72631fc6c0eb6dde34a73e44b"
dependencies = [
"num-traits",
"pxfm",
]
[[package]]
name = "muda"
version = "0.19.3"
@@ -3606,15 +3478,6 @@ dependencies = [
"objc2-foundation",
]
[[package]]
name = "object"
version = "0.37.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ff76201f031d8863c38aa7f905eca4f53abbfa15f609db4277d44cd8938f33fe"
dependencies = [
"memchr",
]
[[package]]
name = "once_cell"
version = "1.21.4"
@@ -3930,26 +3793,6 @@ dependencies = [
"siphasher",
]
[[package]]
name = "pin-project"
version = "1.1.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924"
dependencies = [
"pin-project-internal",
]
[[package]]
name = "pin-project-internal"
version = "1.1.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "pin-project-lite"
version = "0.2.17"
@@ -3992,7 +3835,7 @@ dependencies = [
"anyhow",
"async-trait",
"core-api",
"reqwest 0.13.4",
"reqwest",
"serde",
"serde_json",
"tokio",
@@ -4007,48 +3850,12 @@ dependencies = [
"async-trait",
"core-api",
"parking_lot",
"reqwest 0.13.4",
"reqwest",
"serde_json",
"tokio",
"tracing",
]
[[package]]
name = "plugin-honcho"
version = "0.1.0"
dependencies = [
"anyhow",
"async-trait",
"core-api",
"honcho-client",
"serde_json",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "plugin-mobile-connector"
version = "0.1.0"
dependencies = [
"anyhow",
"async-trait",
"axum",
"chrono",
"core-api",
"hex",
"image",
"qrcode",
"rand 0.9.4",
"serde",
"serde_json",
"skald-relay-client",
"skald-relay-common",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "plugin-tailscale-remote"
version = "0.1.0"
@@ -4064,24 +3871,6 @@ dependencies = [
"tracing",
]
[[package]]
name = "plugin-telegram-bot"
version = "0.1.0"
dependencies = [
"anyhow",
"async-trait",
"chrono",
"core-api",
"rand 0.10.1",
"regex",
"serde",
"serde_json",
"teloxide",
"tokio",
"tokio-util",
"tracing",
]
[[package]]
name = "plugin-transcribe-whisper-local"
version = "0.1.0"
@@ -4103,7 +3892,7 @@ dependencies = [
"anyhow",
"async-trait",
"core-api",
"reqwest 0.13.4",
"reqwest",
"serde_json",
"tokio",
"tracing",
@@ -4116,7 +3905,7 @@ dependencies = [
"anyhow",
"async-trait",
"core-api",
"reqwest 0.13.4",
"reqwest",
"serde_json",
"tokio",
"tracing",
@@ -4445,31 +4234,6 @@ version = "3.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "95067976aca6421a523e491fce939a3e65249bac4b977adee0ee9771568e8aa3"
[[package]]
name = "psm"
version = "0.1.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "645dbe486e346d9b5de3ef16ede18c26e6c70ad97418f4874b8b1889d6e761ea"
dependencies = [
"ar_archive_writer",
"cc",
]
[[package]]
name = "pxfm"
version = "0.1.29"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e0c5ccf5294c6ccd63a74f1565028353830a9c2f5eb0c682c355c471726a6e3f"
[[package]]
name = "qrcode"
version = "0.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d68782463e408eb1e668cf6152704bd856c78c5b6417adaee3203d8f4c1fc9ec"
dependencies = [
"image",
]
[[package]]
name = "quick-xml"
version = "0.41.0"
@@ -4479,61 +4243,6 @@ dependencies = [
"memchr",
]
[[package]]
name = "quinn"
version = "0.11.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b9e20a958963c291dc322d98411f541009df2ced7b5a4f2bd52337638cfccf20"
dependencies = [
"bytes",
"cfg_aliases",
"pin-project-lite",
"quinn-proto",
"quinn-udp",
"rustc-hash",
"rustls",
"socket2",
"thiserror 2.0.18",
"tokio",
"tracing",
"web-time",
]
[[package]]
name = "quinn-proto"
version = "0.11.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098"
dependencies = [
"bytes",
"getrandom 0.3.4",
"lru-slab",
"rand 0.9.4",
"ring",
"rustc-hash",
"rustls",
"rustls-pki-types",
"slab",
"thiserror 2.0.18",
"tinyvec",
"tracing",
"web-time",
]
[[package]]
name = "quinn-udp"
version = "0.5.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd"
dependencies = [
"cfg_aliases",
"libc",
"once_cell",
"socket2",
"tracing",
"windows-sys 0.60.2",
]
[[package]]
name = "quote"
version = "1.0.45"
@@ -4647,15 +4356,6 @@ version = "0.6.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "20675572f6f24e9e76ef639bc5552774ed45f1c30e2951e1e99c59888861c539"
[[package]]
name = "rc-box"
version = "1.3.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "897fecc9fac6febd4408f9e935e86df739b0023b625e610e0357535b9c8adad0"
dependencies = [
"erasable",
]
[[package]]
name = "redox_syscall"
version = "0.5.18"
@@ -4742,48 +4442,6 @@ version = "0.8.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a"
[[package]]
name = "reqwest"
version = "0.12.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "eddd3ca559203180a307f12d114c268abf583f59b03cb906fd0b3ff8646c1147"
dependencies = [
"base64 0.22.1",
"bytes",
"futures-core",
"futures-util",
"http",
"http-body",
"http-body-util",
"hyper",
"hyper-rustls",
"hyper-util",
"js-sys",
"log",
"mime_guess",
"percent-encoding",
"pin-project-lite",
"quinn",
"rustls",
"rustls-pki-types",
"serde",
"serde_json",
"serde_urlencoded",
"sync_wrapper",
"tokio",
"tokio-rustls",
"tokio-util",
"tower",
"tower-http 0.6.11",
"tower-service",
"url",
"wasm-bindgen",
"wasm-bindgen-futures",
"wasm-streams 0.4.2",
"web-sys",
"webpki-roots 1.0.7",
]
[[package]]
name = "reqwest"
version = "0.13.4"
@@ -4823,19 +4481,10 @@ dependencies = [
"url",
"wasm-bindgen",
"wasm-bindgen-futures",
"wasm-streams 0.5.0",
"wasm-streams",
"web-sys",
]
[[package]]
name = "rgb"
version = "0.8.53"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47b34b781b31e5d73e9fbc8689c70551fd1ade9a19e3e28cfec8580a79290cc4"
dependencies = [
"bytemuck",
]
[[package]]
name = "ring"
version = "0.17.14"
@@ -4952,7 +4601,6 @@ version = "1.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9"
dependencies = [
"web-time",
"zeroize",
]
@@ -5122,7 +4770,7 @@ checksum = "c5d9c0c92a92d33f08817311cf3f2c29a3538a8240e94a6a3c622ce652d7e00c"
dependencies = [
"bitflags 2.11.1",
"cssparser",
"derive_more 2.1.1",
"derive_more",
"log",
"new_debug_unreachable",
"phf 0.13.1",
@@ -5475,14 +5123,11 @@ dependencies = [
"notify",
"plugin-comfyui",
"plugin-elevenlabs",
"plugin-honcho",
"plugin-mobile-connector",
"plugin-tailscale-remote",
"plugin-telegram-bot",
"plugin-transcribe-whisper-local",
"plugin-tts-kokoro",
"plugin-tts-orpheus-3b",
"reqwest 0.13.4",
"reqwest",
"rustls",
"serde",
"serde_json",
@@ -5529,7 +5174,7 @@ dependencies = [
"quote",
"rand 0.10.1",
"regex",
"reqwest 0.13.4",
"reqwest",
"serde",
"serde_json",
"sha2 0.10.9",
@@ -5623,7 +5268,7 @@ dependencies = [
"jsonwebtoken",
"prost",
"rand 0.8.6",
"reqwest 0.13.4",
"reqwest",
"serde",
"serde_json",
"sha2 0.10.9",
@@ -5931,19 +5576,6 @@ version = "1.2.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
[[package]]
name = "stacker"
version = "0.1.24"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "640c8cdd92b6b12f5bcb1803ca3bbf5ab96e5e6b6b96b9ab77dabe9e880b3190"
dependencies = [
"cc",
"cfg-if",
"libc",
"psm",
"windows-sys 0.61.2",
]
[[package]]
name = "static_assertions"
version = "1.1.0"
@@ -6139,18 +5771,6 @@ dependencies = [
"url",
]
[[package]]
name = "take_mut"
version = "0.2.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f764005d11ee5f36500a149ace24e00e3da98b0158b3e2d53a7495660d3f4d60"
[[package]]
name = "takecell"
version = "0.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "20f34339676cdcab560c9a82300c4c2581f68b9369aedf0fae86f2ff9565ff3e"
[[package]]
name = "tao"
version = "0.35.3"
@@ -6238,7 +5858,7 @@ dependencies = [
"percent-encoding",
"plist",
"raw-window-handle",
"reqwest 0.13.4",
"reqwest",
"serde",
"serde_json",
"serde_repr",
@@ -6421,76 +6041,6 @@ dependencies = [
"toml 1.1.2+spec-1.1.0",
]
[[package]]
name = "teloxide"
version = "0.17.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "84992abeed3ae42e8401b25d266d12bcba1def0abe59d22f6b9781167545f71e"
dependencies = [
"aquamarine",
"bytes",
"derive_more 1.0.0",
"dptree",
"either",
"futures",
"log",
"mime",
"pin-project",
"serde",
"serde_json",
"teloxide-core",
"teloxide-macros",
"thiserror 2.0.18",
"tokio",
"tokio-stream",
"tokio-util",
"url",
]
[[package]]
name = "teloxide-core"
version = "0.13.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7f7a34ca8e971fa892e633858c07547fe138ef4a02e4a4eaa1d35e517d6e0bc4"
dependencies = [
"bitflags 2.11.1",
"bytes",
"chrono",
"derive_more 1.0.0",
"either",
"futures",
"log",
"mime",
"once_cell",
"pin-project",
"rc-box",
"reqwest 0.12.28",
"rgb",
"serde",
"serde_json",
"serde_with",
"stacker",
"take_mut",
"takecell",
"thiserror 2.0.18",
"tokio",
"tokio-util",
"url",
"uuid",
]
[[package]]
name = "teloxide-macros"
version = "0.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "300fadcaf0c182f19b5ca10bf23a45dc9a48925f00c704405fd90ee2c03942f9"
dependencies = [
"heck 0.5.0",
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "tempfile"
version = "3.27.0"
@@ -8004,19 +7554,6 @@ dependencies = [
"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]]
name = "wasm-streams"
version = "0.5.0"
@@ -8052,16 +7589,6 @@ dependencies = [
"wasm-bindgen",
]
[[package]]
name = "web-time"
version = "1.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5a6580f308b1fad9207618087a65c04e7a10bc77e02c8e84e9b00dd4b12fa0bb"
dependencies = [
"js-sys",
"wasm-bindgen",
]
[[package]]
name = "web_atoms"
version = "0.2.5"
Cargo.toml
-6
View File
@@ -8,9 +8,7 @@ members = [
"crates/core-api",
"crates/mcp-client",
"crates/plugin-tailscale-remote",
"crates/plugin-honcho",
"crates/plugin-transcribe-whisper-local",
"crates/plugin-telegram-bot",
"crates/plugin-comfyui",
"crates/plugin-tts-orpheus-3b",
"crates/plugin-tts-kokoro",
@@ -18,7 +16,6 @@ members = [
"crates/skald-relay-common",
"crates/skald-relay-server",
"crates/skald-relay-client",
"crates/plugin-mobile-connector",
]
resolver = "2"
@@ -84,14 +81,11 @@ llm-client = { path = "crates/llm-client" }
core-api = { path = "crates/core-api" }
mcp-client = { path = "crates/mcp-client" }
plugin-tailscale-remote = { path = "crates/plugin-tailscale-remote" }
plugin-honcho = { path = "crates/plugin-honcho" }
plugin-transcribe-whisper-local = { path = "crates/plugin-transcribe-whisper-local", optional = true }
plugin-telegram-bot = { path = "crates/plugin-telegram-bot" }
plugin-comfyui = { path = "crates/plugin-comfyui" }
plugin-tts-orpheus-3b = { path = "crates/plugin-tts-orpheus-3b" }
plugin-tts-kokoro = { path = "crates/plugin-tts-kokoro" }
plugin-elevenlabs = { path = "crates/plugin-elevenlabs" }
plugin-mobile-connector = { path = "crates/plugin-mobile-connector" }
# ── Desktop bundle (Tauri) ───────────────────────────────────────────────────
# Optional, activated by the `desktop` feature. Wraps the headless server in a
crates/core-api/src/chatbot.rs
+7
View File
@@ -39,6 +39,13 @@ pub struct ChatOptions {
/// providers — only the logging wrapper reads them.
pub session_id: Option<i64>,
pub stack_id: Option<i64>,
/// The authenticated user driving this request. Correlates the metadata row
/// in `system.db` with the payload in `{userid}.db`. Logging-only.
pub user_id: Option<String>,
/// UUID correlating the metadata row (`llm_requests`) with the payload row
/// (`llm_request_payloads`). Generated by the LLM loop before the call.
/// Logging-only.
pub request_id: Option<String>,
}
/// Raw HTTP metadata captured during a provider call.
crates/core-api/src/plugin.rs
-6
View File
@@ -6,7 +6,6 @@ use serde_json::Value;
use tokio::sync::RwLock;
use crate::approval::ApprovalApi;
use crate::bus::ChatEventBus;
use crate::command::CommandApi;
use crate::system_bus::SystemEventBus;
use crate::chat_hub::ChatHubApi;
@@ -31,13 +30,9 @@ pub type RouterFactory = Arc<dyn Fn() -> axum::Router + Send + Sync>;
/// `RemotePlugin`.
#[derive(Clone)]
pub struct PluginContext {
pub chat_hub: Arc<dyn ChatHubApi>,
/// Custom file-based slash commands (`commands/<name>/`). Read-only from the
/// plugin side — lets the Telegram bot resolve `/command` expansions.
pub command: Arc<dyn CommandApi>,
pub approval: Arc<dyn ApprovalApi>,
/// Unified Inbox façade (approvals + clarifications). See plugin.md §12.2.
pub inbox: Arc<dyn InboxApi>,
/// Skald's shared SQLite pool — lets plugins create/use their own tables
/// (e.g. `relay_*`) in the main DB. See plugin.md §12.1.
pub db: Arc<sqlx::SqlitePool>,
@@ -49,7 +44,6 @@ pub struct PluginContext {
pub tts_provider: Arc<dyn TtsProvider>,
pub api_provider_registry: Arc<dyn ApiProviderRegistry>,
pub location: Arc<dyn LocationUpdater>,
pub event_bus: Arc<ChatEventBus>,
pub system_bus: Arc<SystemEventBus>,
pub web_port: u16,
pub remote_slot: Arc<RwLock<Option<Arc<dyn RemoteAccess>>>>,
crates/core-api/src/tool.rs
+28 -1
View File
@@ -1,6 +1,6 @@
use std::future::Future;
use std::pin::Pin;
use std::sync::Mutex;
use std::sync::{Arc, Mutex};
use anyhow::Result;
use serde_json::Value;
@@ -34,6 +34,24 @@ pub enum ToolCategory {
Config,
}
// ── ToolContext ───────────────────────────────────────────────────────────────
/// Per-invocation execution context threaded into a tool's [`Tool::run_with`].
///
/// Carries the identity of the session driving the call and the owner's database
/// pool. Owner-bound tools (e.g. cron management) read `pool` to act on the
/// caller's own `{userid}.db` rather than a manager captured globally at
/// registration time. Context-free tools ignore it — the default `run_with`
/// delegates to `run`, so most tools need no change.
#[derive(Clone)]
pub struct ToolContext {
/// The session that issued this tool call. Ids are local to `pool`.
pub session_id: i64,
/// The owner's unlocked database pool (per-user in multi-user mode; the shared
/// `system.db` in the transitional single-pool state).
pub pool: Arc<sqlx::SqlitePool>,
}
// ── Tool trait ────────────────────────────────────────────────────────────────
/// A single LLM-callable tool.
@@ -91,6 +109,15 @@ pub trait Tool: Send + Sync {
Box::new(SimpleExecution::new(self.execute_typed(args)))
}
/// Context-aware variant of [`run`](Self::run): the session driver threads a
/// [`ToolContext`] (session id + owner pool) so owner-bound tools can act on
/// the caller's own database instead of a globally-captured manager. The
/// default ignores the context and delegates to `run`, so context-free tools
/// need no change.
fn run_with<'a>(&'a self, _ctx: &ToolContext, args: Value) -> Box<dyn ToolExecution + 'a> {
self.run(args)
}
/// Logical category of this tool.
fn category(&self) -> ToolCategory;
crates/skald-core/src/chatbot/logging.rs
+29 -53
View File
@@ -1,11 +1,12 @@
//! Transparent logging wrapper for any [`ChatbotClient`].
//!
//! [`LoggingChatbotClient`] intercepts every `chat_with_tools` call, captures
//! the raw HTTP request/response from the inner provider via `chat_with_tools_raw`,
//! then persists a row to `llm_requests` asynchronously (fire-and-forget).
//! [`LoggingChatbotClient`] intercepts every `chat_with_tools_raw` call, captures
//! the raw HTTP request/response from the inner provider, persists a **metadata-only**
//! row to `llm_requests` in `system.db` (fire-and-forget), then returns the raw data
//! to the caller so it can write the **payload** to the user's own database.
//!
//! The LLM loop is completely unaware of this: it only holds an
//! `Arc<dyn ChatbotClient>` and calls `chat_with_tools` as usual.
//! The split keeps conversation content (payloads) behind the user key while
//! metadata (cost, tokens, timing) stays in the admin-readable registry.
use std::sync::Arc;
use std::time::Instant;
@@ -21,26 +22,10 @@ use super::{ChatOptions, ChatResponse, ChatbotClient, LlmRawMeta, LlmTurn, Messa
// ─────────────────────────────────────────────────────────────────────────────
/// Controls which parts of the HTTP exchange are persisted per row.
#[derive(Debug, Clone, Copy)]
pub struct LogSaveFlags {
pub request_payload: bool,
pub response_payload: bool,
pub request_headers: bool,
pub response_headers: bool,
}
impl Default for LogSaveFlags {
fn default() -> Self {
Self { request_payload: true, response_payload: true, request_headers: true, response_headers: true }
}
}
pub struct LoggingChatbotClient {
inner: Arc<dyn ChatbotClient>,
pool: Arc<SqlitePool>,
model_name: String,
flags: LogSaveFlags,
}
impl LoggingChatbotClient {
@@ -48,9 +33,8 @@ impl LoggingChatbotClient {
inner: Arc<dyn ChatbotClient>,
pool: Arc<SqlitePool>,
model_name: impl Into<String>,
flags: LogSaveFlags,
) -> Self {
Self { inner, pool, model_name: model_name.into(), flags }
Self { inner, pool, model_name: model_name.into() }
}
}
@@ -65,20 +49,35 @@ impl ChatbotClient for LoggingChatbotClient {
self.inner.chat(messages, options).await
}
/// Intercepts the call, delegates to `inner.chat_with_tools_raw` to capture
/// HTTP wire data, then spawns a fire-and-forget DB write before returning.
/// Passthrough that drops the raw meta. Used by callers that do not need
/// payload capture (e.g. the compactor).
async fn chat_with_tools(
&self,
messages: &[Value],
tools: &[Value],
options: &ChatOptions,
) -> anyhow::Result<LlmTurn> {
let (turn, _) = self.chat_with_tools_raw(messages, tools, options).await?;
Ok(turn)
}
/// Intercepts the call, delegates to `inner.chat_with_tools_raw` to capture
/// HTTP wire data, writes a **metadata-only** row to `system.db`, then returns
/// the raw data so the caller can persist payloads to the user's own database.
async fn chat_with_tools_raw(
&self,
messages: &[Value],
tools: &[Value],
options: &ChatOptions,
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
let start = Instant::now();
let result = self.inner.chat_with_tools_raw(messages, tools, options).await;
let duration_ms = start.elapsed().as_millis() as i64;
let session_id = options.session_id;
let stack_id = options.stack_id;
let user_id = options.user_id.clone();
let request_id = options.request_id.clone();
let model_name = self.model_name.clone();
let pool = Arc::clone(&self.pool);
@@ -90,24 +89,13 @@ impl ChatbotClient for LoggingChatbotClient {
(*input_tokens, *output_tokens, *cache_read_tokens, *cache_creation_tokens),
};
let meta = meta.unwrap_or_default();
let flags = self.flags;
let request_json = if flags.request_payload {
meta.request_body.map(|v| v.to_string()).unwrap_or_default()
} else { String::new() };
let request_headers = if flags.request_headers { meta.request_headers.map(|v| v.to_string()) } else { None };
let response_json = if flags.response_payload { meta.response_body.map(|v| v.to_string()) } else { None };
let response_headers = if flags.response_headers { meta.response_headers.map(|v| v.to_string()) } else { None };
tokio::spawn(async move {
if let Err(e) = llm_requests::insert(&pool, llm_requests::LlmRequestRow {
request_id,
user_id,
session_id,
stack_id,
model_name,
request_json,
request_headers,
response_json,
response_headers,
error_text: None,
input_tokens: input_tokens.map(|n| n as i64),
output_tokens: output_tokens.map(|n| n as i64),
@@ -119,7 +107,7 @@ impl ChatbotClient for LoggingChatbotClient {
}
});
Ok(turn)
Ok((turn, meta))
}
Err(e) => {
@@ -127,13 +115,11 @@ impl ChatbotClient for LoggingChatbotClient {
tokio::spawn(async move {
if let Err(log_err) = llm_requests::insert(&pool, llm_requests::LlmRequestRow {
request_id,
user_id,
session_id,
stack_id,
model_name,
request_json: String::new(),
request_headers: None,
response_json: None,
response_headers: None,
error_text: Some(error_text),
input_tokens: None,
output_tokens: None,
@@ -149,14 +135,4 @@ impl ChatbotClient for LoggingChatbotClient {
}
}
}
/// Expose raw metadata so this wrapper can itself be wrapped if needed.
async fn chat_with_tools_raw(
&self,
messages: &[Value],
tools: &[Value],
options: &ChatOptions,
) -> anyhow::Result<(LlmTurn, Option<LlmRawMeta>)> {
self.inner.chat_with_tools_raw(messages, tools, options).await
}
}
crates/skald-core/src/compactor.rs
+2
View File
@@ -313,6 +313,8 @@ impl ContextCompactor {
temperature: Some(0.3),
session_id: Some(session_id),
stack_id: Some(stack_id),
user_id: None,
request_id: None,
};
let turn = llm.client.chat_with_tools(&messages_payload, &[], &options).await
crates/skald-core/src/db/llm_request_payloads.rs
+32
View File
@@ -0,0 +1,32 @@
//! DB operations for the `llm_request_payloads` table (owner bucket).
//!
//! Full request/response payloads + headers for each LLM call. Lives in
//! `{userid}.db` (encrypted), correlated with the metadata row in `system.db`
//! via `request_id`.
use anyhow::Result;
use sqlx::SqlitePool;
pub struct PayloadRow {
pub request_json: String,
pub request_headers: Option<String>,
pub response_json: Option<String>,
pub response_headers: Option<String>,
pub request_id: String,
}
pub async fn insert(pool: &SqlitePool, row: PayloadRow) -> Result<()> {
sqlx::query(
"INSERT INTO llm_request_payloads
(request_id, request_json, request_headers, response_json, response_headers)
VALUES (?, ?, ?, ?, ?)",
)
.bind(&row.request_id)
.bind(&row.request_json)
.bind(&row.request_headers)
.bind(&row.response_json)
.bind(&row.response_headers)
.execute(pool)
.await?;
Ok(())
}
crates/skald-core/src/db/llm_requests/cleanup.rs
+5 -26
View File
@@ -1,9 +1,9 @@
//! Background maintenance task for the `llm_requests` table.
//!
//! Periodically nulls out old payloads/headers and deletes expired rows according
//! to the retention settings in [`LlmRequestsLogConfig`], then `VACUUM`s to reclaim
//! freed pages. Extracted from `Skald::new` so the loop lives next to the queries it
//! calls; the returned handle is registered with the `TaskSupervisor` for shutdown.
//! Deletes expired metadata rows according to the retention settings in
//! [`LlmRequestsLogConfig`], then `VACUUM`s to reclaim freed pages.
//! Payload/header nulling is gone — those columns moved to `llm_request_payloads`
//! in the owner bucket.
use std::sync::Arc;
use std::time::Duration;
@@ -31,27 +31,6 @@ pub fn spawn(
_ = tokio::time::sleep(Duration::from_secs(60)) => {}
}
loop {
if let Some(days) = cfg.cleanup_request_payload_after {
match super::null_request_payload(&pool, days).await {
Ok(n) if n > 0 => info!(rows = n, days, "llm_requests: nulled request payload"),
Ok(_) => {}
Err(e) => warn!(error = %e, "llm_requests: null request payload failed"),
}
}
if let Some(days) = cfg.cleanup_response_payload_after {
match super::null_response_payload(&pool, days).await {
Ok(n) if n > 0 => info!(rows = n, days, "llm_requests: nulled response payload"),
Ok(_) => {}
Err(e) => warn!(error = %e, "llm_requests: null response payload failed"),
}
}
if let Some(days) = cfg.cleanup_headers_after {
match super::null_headers(&pool, days).await {
Ok(n) if n > 0 => info!(rows = n, days, "llm_requests: nulled headers"),
Ok(_) => {}
Err(e) => warn!(error = %e, "llm_requests: null headers failed"),
}
}
if let Some(days) = cfg.cleanup_rows_after {
match super::delete_old_rows(&pool, days).await {
Ok(n) if n > 0 => info!(deleted = n, days, "llm_requests: deleted old rows"),
@@ -59,7 +38,7 @@ pub fn spawn(
Err(e) => warn!(error = %e, "llm_requests: delete old rows failed"),
}
}
// VACUUM reclaims pages freed by DELETE/UPDATE NULL.
// VACUUM reclaims pages freed by DELETE.
match sqlx::query("VACUUM").execute(&*pool).await {
Ok(_) => info!("llm_requests: VACUUM complete"),
Err(e) => warn!(error = %e, "llm_requests: VACUUM failed"),
crates/skald-core/src/db/llm_requests/mod.rs
+9 -61
View File
@@ -1,7 +1,9 @@
//! DB operations for the `llm_requests` table.
//! DB operations for the `llm_requests` table (metadata only).
//!
//! Every `chat_with_tools` call is logged here by the
//! [`crate::chatbot::logging::LoggingChatbotClient`] wrapper.
//! Payloads (request/response bodies + headers) live in `llm_request_payloads`
//! in the owner bucket (`{userid}.db`), correlated by `request_id`.
//! Rows are retained for `llm.request_log.retention_days` days (default 14).
use anyhow::Result;
@@ -12,17 +14,11 @@ pub mod cleanup;
// ── Row struct ────────────────────────────────────────────────────────────────
pub struct LlmRequestRow {
pub request_id: Option<String>,
pub user_id: Option<String>,
pub session_id: Option<i64>,
pub stack_id: Option<i64>,
pub model_name: String,
/// Full HTTP request body sent to the provider (compact JSON, no pretty-print).
pub request_json: String,
/// HTTP request headers as a compact JSON object (api-key redacted).
pub request_headers: Option<String>,
/// Full HTTP response body from the provider (compact JSON).
pub response_json: Option<String>,
/// HTTP response headers as a compact JSON object.
pub response_headers: Option<String>,
/// Error message when the HTTP call itself failed (no response available).
pub error_text: Option<String>,
pub input_tokens: Option<i64>,
@@ -40,21 +36,17 @@ pub struct LlmRequestRow {
pub async fn insert(pool: &SqlitePool, row: LlmRequestRow) -> Result<i64> {
let id = sqlx::query_scalar::<_, i64>(
"INSERT INTO llm_requests (
session_id, stack_id, model_name,
request_json, request_headers,
response_json, response_headers,
request_id, user_id, session_id, stack_id, model_name,
error_text, input_tokens, output_tokens, duration_ms,
cache_read_tokens, cache_creation_tokens
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
RETURNING id",
)
.bind(&row.request_id)
.bind(&row.user_id)
.bind(row.session_id)
.bind(row.stack_id)
.bind(&row.model_name)
.bind(&row.request_json)
.bind(&row.request_headers)
.bind(&row.response_json)
.bind(&row.response_headers)
.bind(&row.error_text)
.bind(row.input_tokens)
.bind(row.output_tokens)
@@ -79,47 +71,3 @@ pub async fn delete_old_rows(pool: &SqlitePool, days: u32) -> Result<u64> {
.rows_affected();
Ok(n)
}
/// Nulls out `request_json` for rows older than `days` days. Returns rows affected.
pub async fn null_request_payload(pool: &SqlitePool, days: u32) -> Result<u64> {
let cutoff = format!("-{days} days");
let n = sqlx::query(
"UPDATE llm_requests SET request_json = '' \
WHERE request_json != '' AND created_at < datetime('now', ?)",
)
.bind(&cutoff)
.execute(pool)
.await?
.rows_affected();
Ok(n)
}
/// Nulls out `response_json` for rows older than `days` days. Returns rows affected.
pub async fn null_response_payload(pool: &SqlitePool, days: u32) -> Result<u64> {
let cutoff = format!("-{days} days");
let n = sqlx::query(
"UPDATE llm_requests SET response_json = NULL \
WHERE response_json IS NOT NULL AND created_at < datetime('now', ?)",
)
.bind(&cutoff)
.execute(pool)
.await?
.rows_affected();
Ok(n)
}
/// Nulls out both header columns for rows older than `days` days. Returns rows affected.
pub async fn null_headers(pool: &SqlitePool, days: u32) -> Result<u64> {
let cutoff = format!("-{days} days");
let n = sqlx::query(
"UPDATE llm_requests \
SET request_headers = NULL, response_headers = NULL \
WHERE (request_headers IS NOT NULL OR response_headers IS NOT NULL) \
AND created_at < datetime('now', ?)",
)
.bind(&cutoff)
.execute(pool)
.await?
.rows_affected();
Ok(n)
}
crates/skald-core/src/db/mod.rs
+24 -4
View File
@@ -10,6 +10,7 @@ pub mod config;
pub mod job_runs;
pub mod known_tools;
pub mod llm_requests;
pub mod llm_request_payloads;
pub mod mcp_events;
pub mod mcp_servers;
pub mod plugins;
@@ -317,16 +318,17 @@ async fn create_registry_tables(pool: &SqlitePool) -> Result<()> {
// decrypting anything: the admin sees how much, when and which model — never
// what was said. `session_id` / `stack_id` are bare integers, not foreign
// keys, precisely because the rows they point at live in another file.
// `user_id` correlates the row with the payload in `{userid}.db`.
// Payloads (request/response bodies, headers) live in `llm_request_payloads`
// in the owner bucket — they are conversation content, behind the user key.
sqlx::query(
"CREATE TABLE IF NOT EXISTS llm_requests (
id INTEGER PRIMARY KEY AUTOINCREMENT,
request_id TEXT,
user_id TEXT,
session_id INTEGER,
stack_id INTEGER,
model_name TEXT NOT NULL,
request_json TEXT NOT NULL DEFAULT '',
request_headers TEXT,
response_json TEXT,
response_headers TEXT,
error_text TEXT,
input_tokens INTEGER,
output_tokens INTEGER,
@@ -691,6 +693,23 @@ pub async fn create_owner_tables(pool: &SqlitePool) -> Result<()> {
.execute(pool)
.await?;
// Full request/response payloads for telemetry. Lives in the owner bucket
// (per-user, encrypted) because it is conversation content. Correlated with
// the metadata row in `system.db` via `request_id` (uuid).
sqlx::query(
"CREATE TABLE IF NOT EXISTS llm_request_payloads (
id INTEGER PRIMARY KEY AUTOINCREMENT,
request_id TEXT NOT NULL,
request_json TEXT NOT NULL DEFAULT '',
request_headers TEXT,
response_json TEXT,
response_headers TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)",
)
.execute(pool)
.await?;
Ok(())
}
@@ -744,6 +763,7 @@ mod tests {
one("INSERT INTO secrets (key, value) VALUES ('k', 'v')").await.unwrap();
one("INSERT INTO projects (id, name, path) VALUES (1, 'p', '/tmp')").await.unwrap();
one("INSERT INTO project_tickets (project_id, title, job_id) VALUES (1, 't', 1)").await.unwrap();
one("INSERT INTO llm_request_payloads (request_id, request_json) VALUES ('r1', '{}')").await.unwrap();
pool.close().await;
let _ = std::fs::remove_dir_all(&dir);
crates/skald-core/src/db/users.rs
+2
View File
@@ -456,6 +456,8 @@ mod tests {
async fn cleartext_user_round_trips_with_and_without_a_verifier() {
let path = temp_db_path("users-clear");
let pool = crate::db::init_system_pool(&path).await.unwrap();
crate::db::roles::insert(&pool, "children", "Children", "default", None)
.await.unwrap();
insert(&pool, "u-1", "kid", None, "children", &cleartext()).await.unwrap();
insert(&pool, "u-2", "kiosk", None, "children", &Credentials::Cleartext(None)).await.unwrap();
crates/skald-core/src/llm/manager.rs
+10 -10
View File
@@ -9,7 +9,7 @@ use tokio::sync::RwLock;
use tracing::{info, warn};
use crate::chatbot::ChatbotClient;
use crate::chatbot::logging::{LoggingChatbotClient, LogSaveFlags};
use crate::chatbot::logging::LoggingChatbotClient;
use core_api::provider::LlmStrength;
use crate::provider::{ApiProvider, ProviderRegistry, ReasoningMode};
@@ -69,15 +69,15 @@ pub struct LlmManager {
catalog: RwLock<HashMap<i64, CachedCatalog>>,
/// Per-model metadata cache, keyed by model display name. TTL = 1h.
model_meta_cache: RwLock<HashMap<String, CachedModelMeta>>,
/// When `Some`, every LLM entry is wrapped with [`LoggingChatbotClient`].
log_flags: Option<LogSaveFlags>,
/// When `true`, every LLM entry is wrapped with [`LoggingChatbotClient`].
log_enabled: bool,
}
impl LlmManager {
pub async fn new(
pool: Arc<SqlitePool>,
registry: Arc<ProviderRegistry>,
log_flags: Option<LogSaveFlags>,
log_enabled: bool,
) -> Result<Arc<Self>> {
let mgr = Arc::new(Self {
pool,
@@ -89,7 +89,7 @@ impl LlmManager {
}),
catalog: RwLock::new(HashMap::new()),
model_meta_cache: RwLock::new(HashMap::new()),
log_flags,
log_enabled,
});
mgr.reload().await?;
Ok(mgr)
@@ -461,9 +461,9 @@ impl LlmManager {
}
};
let log_config = self.log_flags.map(|f| (Arc::clone(&self.pool), f));
let log_pool = self.log_enabled.then(|| Arc::clone(&self.pool));
let entry = match build_entry(&self.registry, &provider, &model, model.id, log_config) {
let entry = match build_entry(&self.registry, &provider, &model, model.id, log_pool) {
Ok(e) => Arc::new(e),
Err(e) => {
warn!(model = %model.name, error = %e, "failed to build LLM entry, skipping");
@@ -505,7 +505,7 @@ fn build_entry(
provider: &LlmProviderRecord,
model: &LlmModelRecord,
model_db_id: i64,
log_config: Option<(Arc<SqlitePool>, LogSaveFlags)>,
log_pool: Option<Arc<SqlitePool>>,
) -> Result<LlmEntry> {
let built = registry.get(&provider.provider)
.ok_or_else(|| anyhow::anyhow!("unknown provider type '{}'", provider.provider))?
@@ -516,8 +516,8 @@ fn build_entry(
let prompt_cache = built.prompt_cache;
let extra = model.extra_params.clone();
let client: Arc<dyn ChatbotClient> = match log_config {
Some((pool, flags)) => Arc::new(LoggingChatbotClient::new(inner, pool, &model.name, flags)),
let client: Arc<dyn ChatbotClient> = match log_pool {
Some(pool) => Arc::new(LoggingChatbotClient::new(inner, pool, &model.name)),
None => inner,
};
crates/skald-core/src/plugin/mod.rs
-4
View File
@@ -96,10 +96,7 @@ impl PluginManager {
.ok_or_else(|| anyhow::anyhow!("PluginManager: web_port not set"))?;
Ok(PluginContext {
chat_hub: Arc::clone(skald.chat_hub()) as _,
command: Arc::clone(skald.command_manager()) as _,
approval: Arc::clone(skald.approval()) as _,
inbox: Arc::new(skald.inbox().clone()) as _,
db: Arc::clone(skald.db()),
secrets: Arc::clone(skald.secrets()) as _,
transcribe: Arc::clone(skald.transcribe_manager()) as _,
@@ -109,7 +106,6 @@ impl PluginManager {
tts_provider: Arc::clone(skald.tts_manager()) as _,
api_provider_registry: Arc::clone(skald.provider_registry()) as _,
location: Arc::clone(skald.location_manager()) as _,
event_bus: Arc::clone(skald.event_bus()),
system_bus: Arc::clone(skald.system_bus()),
web_port,
remote_slot: Arc::clone(skald.remote()),
crates/skald-core/src/session/handler/llm_call.rs
+27 -3
View File
@@ -13,6 +13,7 @@ use tokio_util::sync::CancellationToken;
use tracing::{error, warn};
use crate::chatbot::{ChatOptions, LlmTurn};
use crate::db::llm_request_payloads;
use crate::llm::{LlmEntry, LlmStrength};
use super::ChatSessionHandler;
@@ -54,12 +55,15 @@ impl ChatSessionHandler {
let mut tried_this_round: Vec<String> = vec![cur_name.clone()];
loop {
let request_id = uuid::Uuid::new_v4().to_string();
let options = ChatOptions {
model: cur_llm.model.clone(),
max_tokens: None,
temperature: None,
session_id: Some(self.session_id),
stack_id: Some(stack_id),
user_id: Some(self.user_id.clone()),
request_id: Some(request_id.clone()),
};
// Clone the Arc so the in-flight future does not borrow `cur_llm` across
@@ -68,13 +72,33 @@ impl ChatSessionHandler {
let client = cur_llm.client.clone();
let call_result = tokio::select! {
_ = token.cancelled() => return RoundLlm::Cancelled,
r = client.chat_with_tools(messages.as_slice(), tool_defs, &options) => r,
r = client.chat_with_tools_raw(messages.as_slice(), tool_defs, &options) => r,
};
let e = match call_result {
Ok(t) => {
Ok((turn, meta)) => {
self.llm_manager.mark_success(cur_name).await;
return RoundLlm::Turn(t);
// Persist the payload (request/response bodies + headers) to the
// user's own database. Fire-and-forget — a failed write must not
// break the turn. The metadata row is already written by the
// logging wrapper to system.db with the same request_id.
if let Some(meta) = meta {
let pool = Arc::clone(&self.db);
let rid = request_id.clone();
tokio::spawn(async move {
let row = llm_request_payloads::PayloadRow {
request_id: rid,
request_json: meta.request_body.map(|v| v.to_string()).unwrap_or_default(),
request_headers: meta.request_headers.map(|v| v.to_string()),
response_json: meta.response_body.map(|v| v.to_string()),
response_headers: meta.response_headers.map(|v| v.to_string()),
};
if let Err(e) = llm_request_payloads::insert(&pool, row).await {
tracing::warn!(error = %e, "llm_request_payloads: failed to insert");
}
});
}
return RoundLlm::Turn(turn);
}
Err(e) => e,
};
crates/skald-core/src/session/handler/llm_loop.rs
+5 -2
View File
@@ -9,7 +9,7 @@ use crate::chatbot::{LlmTurn, ToolCall};
use crate::db::{chat_history, chat_llm_tools};
use crate::events::ServerEvent;
use crate::tools::{
ExecutionOutcome, SimpleExecution, ToolDescriptionLength, ToolExecution, ToolResult,
ExecutionOutcome, SimpleExecution, ToolContext, ToolDescriptionLength, ToolExecution, ToolResult,
};
use futures::stream::{self, StreamExt};
@@ -426,6 +426,9 @@ impl ChatSessionHandler {
}
// Built-in registry tools (incl. execute_cmd, whose SimpleExecution kills
// the child via kill_on_drop when the work future is dropped on /stop).
self.tools.run(name, args)
// The ToolContext carries this session's id and owner pool so owner-bound
// registry tools (e.g. cron management) act on the caller's own database.
let ctx = ToolContext { session_id: self.session_id, pool: Arc::clone(&self.db) };
self.tools.run(name, &ctx, args)
}
}
crates/skald-core/src/session/handler/mod.rs
+5
View File
@@ -262,6 +262,9 @@ impl ApprovalDecision {
pub struct ChatSessionHandler {
pub session_id: i64,
pub(super) db: Arc<SqlitePool>,
/// The authenticated user who owns this session. Threaded into `ChatOptions`
/// so the telemetry metadata row in `system.db` carries `user_id`.
pub(super) user_id: String,
pub(super) llm_manager: Arc<LlmManager>,
pub(super) max_history_messages: usize,
pub(super) max_tool_rounds: usize,
@@ -329,6 +332,7 @@ impl ChatSessionHandler {
pub fn new(
session_id: i64,
db: Arc<SqlitePool>,
user_id: String,
llm_manager: Arc<LlmManager>,
max_history_messages: usize,
max_tool_rounds: usize,
@@ -353,6 +357,7 @@ impl ChatSessionHandler {
Self {
session_id,
db,
user_id,
llm_manager,
max_history_messages,
max_tool_rounds,
crates/skald-core/src/session/manager.rs
+4
View File
@@ -22,6 +22,7 @@ use super::handler::ChatSessionHandler;
pub struct ChatSessionManager {
db: Arc<SqlitePool>,
user_id: String,
llm_manager: Arc<LlmManager>,
max_history_messages: usize,
max_tool_rounds: usize,
@@ -47,6 +48,7 @@ pub struct ChatSessionManager {
impl ChatSessionManager {
pub fn new(
db: Arc<SqlitePool>,
user_id: String,
llm_manager: Arc<LlmManager>,
max_history_messages: usize,
max_tool_rounds: usize,
@@ -66,6 +68,7 @@ impl ChatSessionManager {
) -> Self {
Self {
db,
user_id,
llm_manager,
max_history_messages,
max_tool_rounds,
@@ -152,6 +155,7 @@ impl ChatSessionManager {
let handler = Arc::new(ChatSessionHandler::new(
session_id,
self.db.clone(),
self.user_id.clone(),
Arc::clone(&self.llm_manager),
self.max_history_messages,
self.max_tool_rounds,
crates/skald-core/src/skald/accessors.rs
+11
View File
@@ -52,6 +52,17 @@ impl Skald {
// Runtime / cross-cutting
pub fn db(&self) -> &Arc<SqlitePool> { &self.rt.db }
pub fn users(&self) -> &Arc<UserManager> { &self.rt.users }
/// The caller's per-user owner-bound runtime (chat/hub/cron/interaction),
/// built lazily on first use. `None` when the user's database is still locked
/// (not logged in). The pool is the unlock token (§9); a present pool means an
/// unlocked database, so a context can be built for it.
pub async fn user_context(&self, user_id: &str) -> Option<Arc<super::UserContext>> {
let pool = self.rt.users.pool_of(user_id)?;
self.rt_user_contexts().resolve(user_id, pool).await.ok()
}
fn rt_user_contexts(&self) -> &super::user_context::UserContextRegistry { &self.user_contexts }
pub fn sessions(&self) -> &Arc<crate::auth::SessionStore> { &self.rt.sessions }
pub fn config(&self) -> &Arc<GlobalConfigManager> { &self.rt.config }
pub fn config_properties(&self) -> &[core_api::ConfigSet] { &self.rt.config_properties }
crates/skald-core/src/skald/bundles.rs
+4 -11
View File
@@ -72,16 +72,8 @@ impl Models {
let provider_registry = Arc::new(provider_registry);
info!("provider registry ready ({} built-in providers)", provider_registry.all().len());
let log_flags = config.llm.requests_log.as_ref().filter(|r| r.enabled).map(|r| {
use crate::chatbot::logging::LogSaveFlags;
LogSaveFlags {
request_payload: r.request_payload_save,
response_payload: r.response_payload_save,
request_headers: r.request_header_save,
response_headers: r.response_header_save,
}
});
let llm_manager = LlmManager::new(Arc::clone(&rt.db), Arc::clone(&provider_registry), log_flags).await?;
let log_enabled = config.llm.requests_log.as_ref().is_some_and(|r| r.enabled);
let llm_manager = LlmManager::new(Arc::clone(&rt.db), Arc::clone(&provider_registry), log_enabled).await?;
let client_count = llm_manager.client_names().await.len().saturating_sub(1);
let default_client = llm_manager.default_name().await;
info!(clients = client_count, default = %default_client, "LLM clients loaded");
@@ -226,7 +218,7 @@ impl Tools {
Arc::clone(&integrations.mcp), Arc::clone(&integrations.plugin_manager), Arc::clone(&tasks.cron)));
tool_registry.register(crate::tools::register_mcp::RegisterMcp::new(Arc::clone(&integrations.mcp)));
tool_registry.register(crate::tools::register_mcp::DeleteMcp::new(Arc::clone(&integrations.mcp)));
tool_registry.register(crate::tools::cron_jobs::DeleteCronJob(Arc::clone(&tasks.cron)));
tool_registry.register(crate::tools::cron_jobs::DeleteCronJob);
tool_registry.register(crate::tools::set_secret::SetSecret(Arc::clone(&models.secrets)));
tool_registry.register(crate::tools::list_secrets::ListSecrets(Arc::clone(&models.secrets)));
tool_registry.register(crate::tools::configure_plugin::ConfigurePlugin(Arc::clone(&integrations.plugin_manager)));
@@ -346,6 +338,7 @@ impl Conversation {
let manager = Arc::new(ChatSessionManager::new(
Arc::clone(&rt.db),
String::new(),
Arc::clone(&models.llm_manager),
config.llm.max_history_messages,
config.llm.max_tool_rounds.unwrap_or(DEFAULT_MAX_TOOL_ROUNDS),
crates/skald-core/src/skald/mod.rs
+14
View File
@@ -23,10 +23,13 @@ mod accessors;
mod bundles;
mod runtime;
mod supervisor;
mod user_context;
mod wiring;
use bundles::{Conversation, Infra, Integrations, Interaction, Media, Models, Tasks, Tools};
use runtime::Runtime;
use user_context::{UserContextFactory, UserContextRegistry};
pub use user_context::UserContext;
use wiring::{spawn_background, wire};
pub struct Skald {
@@ -39,6 +42,10 @@ pub struct Skald {
conversation: Conversation,
interaction: Interaction,
infra: Infra,
/// Per-user owner-bound runtimes (chat/hub/cron/interaction), built lazily on
/// first use after a user's pool is unlocked. The global bundles above still
/// serve deferred subsystems and the not-yet-migrated call sites.
user_contexts: UserContextRegistry,
}
impl Skald {
@@ -68,8 +75,15 @@ impl Skald {
wire(&tasks, &conversation, &integrations, &interaction);
spawn_background(&rt, &tasks, &conversation, &integrations, config);
// Per-user context factory: captures the global capability managers, so a
// per-user chat/hub/cron/interaction stack can be stamped out on demand.
let user_contexts = UserContextRegistry::new(UserContextFactory::new(
&rt, &models, &media, &tools, &integrations, &conversation, config,
));
let skald = Arc::new(Skald {
rt, models, media, tools, integrations, tasks, conversation, interaction, infra,
user_contexts,
});
// Inject the fully-constructed instance into the plugin manager — the one
crates/skald-core/src/skald/user_context.rs
+259
View File
@@ -0,0 +1,259 @@
//! Per-user runtime context (blueprint §9 / §11 / §5.1).
//!
//! The owner-bound managers — chat sessions, chat hub, cron, and the
//! approval/clarification/elicitation/inbox interaction stack — must operate on
//! *one* user's `{userid}.db` and emit on *one* user's server→client channel, so
//! that no chat content, WS event, job or pending approval crosses between users.
//! They are built **lazily** on first use after the user's pool is unlocked, and
//! live exactly as long as that pool (§9: from first login until restart).
//!
//! `UserManager` stays the pool-lifecycle owner (§11 boundary). This factory sits
//! at the `Skald` layer, where the global *capability* managers (LLM, tools, MCP,
//! memory, providers) are visible, and stamps out the per-user instances against a
//! given pool, wiring their construction cycles and starting the per-user cron
//! loop. Capability managers are shared by reference; only owner-bound state is
//! per-user.
//!
//! Split of pools inside one context: session/history/jobs/hub use the **user
//! pool**; approval *rules* and `known_tools` are instance-wide registry data, so
//! `ApprovalManager` and `ToolDiscovery` read the **registry pool** (`system.db`)
//! while `ApprovalManager` still emits on the user's channel and keeps its own
//! pending map.
use std::collections::HashMap;
use std::sync::Arc;
use anyhow::Result;
use chrono_tz::Tz;
use sqlx::SqlitePool;
use tokio::sync::{broadcast, Mutex};
use tokio_util::sync::CancellationToken;
use core_api::events::GlobalEvent;
use core_api::system_bus::SystemEventBus;
use crate::approval::ApprovalManager;
use crate::chat_event_bus::ChatEventBus;
use crate::chat_hub::ChatHub;
use crate::clarification::ClarificationManager;
use crate::compactor::ContextCompactor;
use crate::config::{CompactionConfig, CoreConfig, DatetimeConfig};
use crate::cron::TaskManager;
use crate::elicitation::ElicitationManager;
use crate::image_generate::ImageGeneratorManager;
use crate::inbox::Inbox;
use crate::llm::LlmManager;
use crate::mcp::McpManager;
use crate::memory::MemoryManager;
use crate::projects::tickets::ProjectTicketManager;
use crate::run_context::RunContextManager;
use crate::session::handler::{DEFAULT_MAX_PARALLEL_SUBAGENTS, DEFAULT_MAX_TOOL_ROUNDS};
use crate::session::manager::ChatSessionManager;
use crate::tool_discovery::ToolDiscovery;
use crate::tools::ToolRegistry;
use super::bundles::{Conversation, Integrations, Media, Models, Tools};
use super::runtime::Runtime;
/// One unlocked user's owner-bound runtime. Lifetime = the pool's lifetime.
pub struct UserContext {
pub user_id: String,
pub pool: Arc<SqlitePool>,
pub event_bus: Arc<ChatEventBus>,
pub sessions: Arc<ChatSessionManager>,
pub chat_hub: Arc<ChatHub>,
pub cron: Arc<TaskManager>,
pub tickets: Arc<ProjectTicketManager>,
pub approval: Arc<ApprovalManager>,
pub clarification: Arc<ClarificationManager>,
pub elicitation: Arc<ElicitationManager>,
pub inbox: Inbox,
/// Per-user server→client push channel. WS handlers subscribe here (via the
/// hub) so a user's `ServerEvent`s never reach another user's socket.
pub global_tx: broadcast::Sender<GlobalEvent>,
}
/// Captures the global capability managers + resolved config once, and stamps out
/// a [`UserContext`] per unlocked pool.
pub(super) struct UserContextFactory {
registry_pool: Arc<SqlitePool>,
llm_manager: Arc<LlmManager>,
tools: Arc<ToolRegistry>,
mcp: Arc<McpManager>,
memory_manager: Arc<MemoryManager>,
image_generator_manager: Arc<ImageGeneratorManager>,
run_context_manager: Arc<RunContextManager>,
system_bus: Arc<SystemEventBus>,
supervisor: Arc<super::supervisor::TaskSupervisor>,
shutdown_token: CancellationToken,
max_history_messages: usize,
max_tool_rounds: usize,
max_parallel_subagents: usize,
max_tool_result_chars: Option<usize>,
datetime_config: DatetimeConfig,
compaction: Option<CompactionConfig>,
cron_tz: Option<Tz>,
}
impl UserContextFactory {
pub(super) fn new(
rt: &Runtime,
models: &Models,
media: &Media,
tools: &Tools,
integrations: &Integrations,
conversation: &Conversation,
config: &CoreConfig,
) -> Self {
let cron_tz = config.timezone.as_deref().and_then(|s| s.parse::<Tz>().ok());
Self {
registry_pool: Arc::clone(&rt.db),
llm_manager: Arc::clone(&models.llm_manager),
tools: Arc::clone(&tools.tools),
mcp: Arc::clone(&integrations.mcp),
memory_manager: Arc::clone(&models.memory_manager),
image_generator_manager: Arc::clone(&media.image_generator_manager),
run_context_manager: Arc::clone(&conversation.run_context_manager),
system_bus: Arc::clone(&rt.system_bus),
supervisor: Arc::clone(&rt.supervisor),
shutdown_token: rt.shutdown_token.clone(),
max_history_messages: config.llm.max_history_messages,
max_tool_rounds: config.llm.max_tool_rounds.unwrap_or(DEFAULT_MAX_TOOL_ROUNDS),
max_parallel_subagents: config.llm.max_parallel_subagents.unwrap_or(DEFAULT_MAX_PARALLEL_SUBAGENTS),
max_tool_result_chars: config.llm.max_tool_result_chars,
datetime_config: DatetimeConfig { timezone: config.timezone.clone(), ..config.llm.datetime },
compaction: config.llm.compaction.clone(),
cron_tz,
}
}
async fn build(&self, user_id: &str, pool: SqlitePool) -> Result<Arc<UserContext>> {
let pool = Arc::new(pool);
let event_bus = Arc::new(ChatEventBus::new());
let (global_tx, _) = broadcast::channel::<GlobalEvent>(512);
// Interaction stack, per-user. Approval reads the shared registry rules but
// emits on this user's channel and keeps its own pending map — no cross-user
// collision on request_id / session_id. Rules are seeded once by the global
// ApprovalManager at boot, so no re-seeding here.
let approval = Arc::new(ApprovalManager::new(Arc::clone(&self.registry_pool), global_tx.clone()));
let clarification = ClarificationManager::new(global_tx.clone());
let elicitation = ElicitationManager::new(global_tx.clone());
let inbox = Inbox::new(
Arc::clone(&approval),
Arc::clone(&clarification),
Arc::clone(&elicitation),
Arc::clone(&self.tools),
);
let compactor = self.compaction.as_ref().map(|cfg| {
Arc::new(ContextCompactor::new(
cfg.clone(),
Arc::clone(&self.llm_manager),
Arc::clone(&event_bus),
))
});
let manager = Arc::new(ChatSessionManager::new(
Arc::clone(&pool),
user_id.to_string(),
Arc::clone(&self.llm_manager),
self.max_history_messages,
self.max_tool_rounds,
self.max_parallel_subagents,
self.max_tool_result_chars,
self.datetime_config.clone(),
Arc::clone(&self.tools),
Arc::clone(&self.mcp),
Arc::clone(&approval),
Arc::clone(&clarification),
Arc::clone(&event_bus),
Arc::clone(&self.memory_manager),
Arc::clone(&self.image_generator_manager),
compactor,
Arc::clone(&self.run_context_manager),
// known_tools is registry data → discovery writes to the registry pool.
Arc::new(ToolDiscovery::new(Arc::clone(&self.registry_pool))),
));
let chat_hub = ChatHub::new(
Arc::clone(&pool),
Arc::clone(&manager),
Arc::clone(&approval),
global_tx.clone(),
self.shutdown_token.clone(),
);
chat_hub.register("web").await;
chat_hub.register("talk").await;
let cron = TaskManager::new(Arc::clone(&pool), self.cron_tz, Arc::clone(&self.system_bus));
cron.set_session(Arc::clone(&manager));
cron.set_hub(Arc::clone(&chat_hub));
cron.set_self_arc(Arc::clone(&cron));
chat_hub.set_task_mgr(Arc::clone(&cron));
// Per-user ticket manager — wired to the per-user TaskManager so
// `start_ticket` spawns jobs in the user's own pool.
let tickets = ProjectTicketManager::new(Arc::clone(&pool));
tickets.set_task_manager(Arc::clone(&cron));
// Per-user cron loop. `start()` observes the shutdown token, so it stops on
// shutdown; adopting it lets the supervisor also join it. The name is leaked
// to satisfy the `&'static str` label — bounded by the (small) user count.
let name: &'static str = Box::leak(format!("cron:{user_id}").into_boxed_str());
self.supervisor.adopt(name, Arc::clone(&cron).start(self.shutdown_token.clone()));
// Per-user ticket-listener: reacts to JobCompleted events for this user's
// tickets. All users' listeners receive the event (global system bus); only
// the one that owns the ticket does the UPDATE — others no-op on 0 rows.
let tname: &'static str = Box::leak(format!("tickets:{user_id}").into_boxed_str());
self.supervisor.adopt_one(
tname,
Arc::clone(&tickets).start_listener(
Arc::clone(&self.system_bus),
self.shutdown_token.clone(),
),
);
Ok(Arc::new(UserContext {
user_id: user_id.to_string(),
pool,
event_bus,
sessions: manager,
chat_hub,
cron,
tickets,
approval,
clarification,
elicitation,
inbox,
global_tx,
}))
}
}
/// The live per-user contexts, keyed by user id, plus the factory that builds them.
/// A `tokio::Mutex` serialises the build so a context (and its cron loop) is created
/// at most once per user, even under concurrent first-use.
pub(super) struct UserContextRegistry {
factory: UserContextFactory,
contexts: Mutex<HashMap<String, Arc<UserContext>>>,
}
impl UserContextRegistry {
pub(super) fn new(factory: UserContextFactory) -> Self {
Self { factory, contexts: Mutex::new(HashMap::new()) }
}
/// Returns the user's context, building it from `pool` on first use. Idempotent:
/// once built, the same `Arc<UserContext>` is returned until restart.
pub(super) async fn resolve(&self, user_id: &str, pool: SqlitePool) -> Result<Arc<UserContext>> {
let mut guard = self.contexts.lock().await;
if let Some(ctx) = guard.get(user_id) {
return Ok(Arc::clone(ctx));
}
let ctx = self.factory.build(user_id, pool).await?;
guard.insert(user_id.to_string(), Arc::clone(&ctx));
Ok(ctx)
}
}
crates/skald-core/src/skald/wiring.rs
+17 -38
View File
@@ -1,6 +1,11 @@
//! Post-construction wiring: the `OnceLock` cycle-breakers and the background-task
//! spawns, each concentrated in one readable place instead of being scattered
//! through the constructor.
//!
//! Owner-bound background loops (cron, session-cancel, ticket-listener, tic) have
//! moved per-user into `UserContextFactory::build`. What remains here are the
//! instance-wide tasks: LLM-log cleanup on the registry pool, and MCP server
//! initialization.
use std::sync::Arc;
@@ -14,6 +19,10 @@ use super::runtime::Runtime;
/// Resolves the construction cycles (`cron ↔ session ↔ hub`, `ticket → cron`,
/// `mcp → elicitation`) via the managers' `OnceLock` setters.
///
/// These wire the **global** bundles, which are transitional and will be removed
/// once all call-sites are per-user (Phase 6). They remain constructed so any
/// not-yet-migrated accessor does not panic on a `None` OnceLock.
pub(super) fn wire(
tasks: &Tasks,
conversation: &Conversation,
@@ -29,14 +38,16 @@ pub(super) fn wire(
info!("ChatHub initialised");
}
/// Spawns every long-lived background task, each registered by name with the
/// supervisor so it is joined on shutdown. MCP `initialize()` is spawned here —
/// after `wire()` has installed the elicitation handler — so stdio servers start
/// with a handler for server-initiated `elicitation/create` requests.
/// Spawns the instance-wide background tasks.
///
/// Owner-bound loops (cron, session-cancel, ticket-listener, tic) are **not**
/// spawned here — they run per-user inside `UserContext`. Session cancellation is
/// handled directly by the API handlers (which have `AuthUser` and resolve the
/// per-user context). TIC is deferred until connectors return (§13).
pub(super) fn spawn_background(
rt: &Runtime,
tasks: &Tasks,
conversation: &Conversation,
_tasks: &Tasks,
_conversation: &Conversation,
integrations: &Integrations,
config: &CoreConfig,
) {
@@ -52,29 +63,6 @@ pub(super) fn spawn_background(
);
}
// Session-cancellation subscriber: fans SessionCancelled events on the system
// bus into cancel_session() so any in-flight turn / approval / clarification
// all unblock.
{
let manager_ref = Arc::clone(&conversation.manager);
let mut rx = rt.system_bus.subscribe();
let sd = rt.shutdown_token.clone();
rt.supervisor.spawn("session-cancel", async move {
loop {
tokio::select! {
_ = sd.cancelled() => break,
event = rx.recv() => match event {
Ok(core_api::system_bus::SystemEvent::SessionCancelled { session_id }) => {
manager_ref.cancel_session(session_id).await;
}
Ok(_) => {}
Err(_) => break,
}
}
}
});
}
// MCP servers connect in the background. `initialize()` does not itself observe
// the cancellation token, so race it against shutdown: on cancel the task exits
// promptly (dropping the in-flight connection attempts) instead of blocking the
@@ -89,13 +77,4 @@ pub(super) fn spawn_background(
}
});
}
rt.supervisor.adopt("cron", Arc::clone(&tasks.cron).start(rt.shutdown_token.clone()));
info!("cron scheduler started");
rt.supervisor.adopt_one(
"ticket-listener",
Arc::clone(&tasks.ticket_manager).start_listener(Arc::clone(&rt.system_bus), rt.shutdown_token.clone()),
);
rt.supervisor.adopt_one("tic", Arc::clone(&conversation.tic_manager).start(rt.shutdown_token.clone()));
info!("TicManager started");
}
crates/skald-core/src/tools/cron_jobs.rs
+15 -7
View File
@@ -4,7 +4,7 @@ use anyhow::Result;
use serde_json::{Value, json};
use crate::cron::TaskManager;
use crate::tools::{Tool, ToolDescriptionLength};
use crate::tools::{SimpleExecution, Tool, ToolContext, ToolDescriptionLength, ToolExecution, ToolResult};
// ── execute_task ──────────────────────────────────────────────────────────────
//
@@ -134,7 +134,11 @@ pub fn build_execute_task_interface_tool(
// ── delete_cron_job ───────────────────────────────────────────────────────────
pub struct DeleteCronJob(pub Arc<TaskManager>);
/// Deleting a job is a plain DELETE on the owner's `scheduled_jobs`, so this tool
/// acts on `ToolContext::pool` (the caller's own database) rather than capturing a
/// globally-scoped `TaskManager` at registration — a job created in a user's own
/// space is deleted from that same space.
pub struct DeleteCronJob;
impl Tool for DeleteCronJob {
fn name(&self) -> &str { "delete_cron_job" }
@@ -159,12 +163,16 @@ impl Tool for DeleteCronJob {
format!("delete cron job #{id}")
}
fn execute(&self, args: Value) -> Result<String> {
fn run_with<'a>(&'a self, ctx: &ToolContext, args: Value) -> Box<dyn ToolExecution + 'a> {
let pool = Arc::clone(&ctx.pool);
Box::new(SimpleExecution::new(Box::pin(async move {
let id = args["id"].as_i64().ok_or_else(|| anyhow::anyhow!("id must be an integer"))?;
if self.0.delete_job(id)? {
Ok(format!("Task {id} deleted."))
let msg = if crate::db::scheduled_jobs::delete(&pool, id).await? {
format!("Task {id} deleted.")
} else {
Ok(format!("No task with id {id}."))
}
format!("No task with id {id}.")
};
Ok(ToolResult::Text(msg))
})))
}
}
crates/skald-core/src/tools/mod.rs
+7 -3
View File
@@ -55,7 +55,7 @@ use anyhow::Result;
use serde_json::Value;
pub use core_api::tool::{
drive_execution, ExecutionOutcome, SimpleExecution, Tool, ToolCategory,
drive_execution, ExecutionOutcome, SimpleExecution, Tool, ToolCategory, ToolContext,
ToolDescriptionLength, ToolExecution, ToolResult, truncate_label,
};
@@ -207,8 +207,12 @@ impl ToolRegistry {
/// Start a [`ToolExecution`] for a registered tool, or `None` if `name` is not
/// in the registry (MCP / interface tools are handled by the caller). The
/// returned handle borrows the registry, which outlives the turn.
pub fn run(&self, name: &str, args: Value) -> Option<Box<dyn ToolExecution + '_>> {
self.tools.get(name).map(|tool| tool.run(args))
///
/// `ctx` carries the caller's session id and owner pool: owner-bound tools
/// (e.g. cron management) act on `ctx.pool` instead of a globally-captured
/// manager. Context-free tools ignore it via the default `run_with`.
pub fn run(&self, name: &str, ctx: &ToolContext, args: Value) -> Option<Box<dyn ToolExecution + '_>> {
self.tools.get(name).map(|tool| tool.run_with(ctx, args))
}
}
crates/skald-core/src/users/mod.rs
+5
View File
@@ -507,6 +507,11 @@ mod tests {
let system = db::init_system_pool(dir.join("system.db").to_str().unwrap())
.await
.unwrap();
// Seed the non-admin role some tests use, so the FK on
// users.role_id → roles(id) holds.
db::roles::insert(&system, "children", "Children", "default", None)
.await
.unwrap();
let users = UserManager {
system: Arc::new(system),
src/frontend/api/agents.rs
+1 -1
View File
@@ -26,7 +26,7 @@ pub async fn get(
) -> Result<Json<AgentDetail>, ApiError> {
let meta = skald_core::agents::load_meta(&id)?;
let prompt = skald_core::agents::load_prompt(&id)?;
let all = skald.manager().llm_manager().list_models_info().await;
let all = skald.llm_manager().list_models_info().await;
let models = sort_models_for_agent(all, meta.scope.as_deref(), meta.strength);
Ok(Json(AgentDetail { meta, prompt, models }))
}
src/frontend/api/approval.rs
+16 -11
View File
@@ -1,24 +1,25 @@
use axum::{
Json,
Json, Extension,
extract::{Path, State},
};
use serde::Deserialize;
use serde_json::{Value, json};
use skald_core::approval::NewApprovalRule;
use skald_core::db::approval_rules;
use skald_core::tool_catalog::{AllTools, McpServerMeta, ToolInfo};
use std::collections::HashSet;
use std::sync::Arc;
use skald_core::skald::Skald;
use super::ApiError;
use super::{ApiError, guard::AuthUser, require_context};
// ── GET /api/approval/rules ───────────────────────────────────────────────────
pub async fn list_rules(
State(skald): State<Arc<Skald>>,
) -> Result<Json<Value>, ApiError> {
let rules = skald.approval().list_rules().await?;
let rules = approval_rules::list(skald.db()).await?;
Ok(Json(json!(rules)))
}
@@ -28,7 +29,7 @@ pub async fn create_rule(
State(skald): State<Arc<Skald>>,
Json(body): Json<NewApprovalRule>,
) -> Result<Json<Value>, ApiError> {
let id = skald.approval().add_rule(body).await?;
let id = approval_rules::insert(skald.db(), body).await?;
Ok(Json(json!({ "id": id })))
}
@@ -42,7 +43,7 @@ pub async fn update_rule(
Path(p): Path<RulePath>,
Json(body): Json<NewApprovalRule>,
) -> Result<Json<Value>, ApiError> {
skald.approval().update_rule(p.id, body).await?;
approval_rules::update(skald.db(), p.id, body).await?;
Ok(Json(json!({ "ok": true })))
}
@@ -52,7 +53,7 @@ pub async fn delete_rule(
State(skald): State<Arc<Skald>>,
Path(p): Path<RulePath>,
) -> Result<Json<Value>, ApiError> {
skald.approval().delete_rule(p.id).await?;
approval_rules::delete(skald.db(), p.id).await?;
Ok(Json(json!({ "ok": true })))
}
@@ -78,14 +79,16 @@ fn default_action() -> String { "approve".to_string() }
pub async fn resolve_pending(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ResolvePath>,
Json(body): Json<ResolveBody>,
) -> Result<Json<Value>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
if body.action == "reject" {
// Pass the raw note; the waiting session builds the canonical message.
skald.inbox().reject(p.request_id, body.note.clone()).await;
ctx.inbox.reject(p.request_id, body.note.clone()).await;
} else {
skald.inbox().approve(p.request_id).await;
ctx.inbox.approve(p.request_id).await;
}
Ok(Json(json!({ "ok": true, "request_id": p.request_id, "action": body.action })))
}
@@ -96,9 +99,11 @@ pub async fn resolve_pending(
pub async fn list_pending(
State(skald): State<Arc<Skald>>,
) -> Json<Value> {
let pending = skald.inbox().list_pending().await.approvals;
Json(json!(pending))
Extension(auth): Extension<AuthUser>,
) -> Result<Json<Value>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let pending = ctx.inbox.list_pending().await.approvals;
Ok(Json(json!(pending)))
}
// ── GET /api/approval/tools ───────────────────────────────────────────────────
src/frontend/api/cron.rs
+32 -15
View File
@@ -1,5 +1,5 @@
use axum::{
Json,
Json, Extension,
extract::{Path, State},
http::StatusCode,
};
@@ -8,7 +8,7 @@ use serde::Deserialize;
use skald_core::db::{scheduled_jobs, job_runs};
use std::sync::Arc;
use skald_core::skald::Skald;
use super::ApiError;
use super::{ApiError, guard::AuthUser, require_context};
#[derive(serde::Serialize)]
pub struct JobResponse {
@@ -29,8 +29,12 @@ pub struct JobResponse {
pub running_since: Option<String>,
}
pub async fn list(State(skald): State<Arc<Skald>>) -> Result<Json<Vec<JobResponse>>, ApiError> {
let jobs = scheduled_jobs::list(skald.db()).await?;
pub async fn list(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
) -> Result<Json<Vec<JobResponse>>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let jobs = scheduled_jobs::list(&ctx.pool).await?;
Ok(Json(jobs.into_iter().map(|j| JobResponse {
id: j.id,
title: j.title,
@@ -51,22 +55,26 @@ pub async fn list(State(skald): State<Arc<Skald>>) -> Result<Json<Vec<JobRespons
}
pub async fn delete_job(
Path(id): Path<i64>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(id): Path<i64>,
) -> Result<(), ApiError> {
let found = scheduled_jobs::delete(skald.db(), id).await?;
let ctx = require_context(&skald, &auth.user_id).await?;
let found = scheduled_jobs::delete(&ctx.pool, id).await?;
if found { Ok(()) } else { Err(ApiError::not_found(format!("job {id} not found"))) }
}
pub async fn toggle(
Path(id): Path<i64>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(id): Path<i64>,
Json(body): Json<serde_json::Value>,
) -> Result<(), ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let enabled = body["enabled"]
.as_bool()
.ok_or_else(|| ApiError::bad_request("'enabled' boolean required"))?;
let found = scheduled_jobs::set_enabled(skald.db(), id, enabled).await?;
let found = scheduled_jobs::set_enabled(&ctx.pool, id, enabled).await?;
if found { Ok(()) } else { Err(ApiError::not_found(format!("job {id} not found"))) }
}
@@ -76,15 +84,17 @@ pub struct SetRunContextBody {
}
pub async fn set_run_context(
Path(id): Path<i64>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(id): Path<i64>,
Json(body): Json<SetRunContextBody>,
) -> Result<(), ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
use skald_core::run_context::RunContext;
let json = body.security_group.as_ref().map(|sg| {
RunContext::with_security_group(Some(sg.clone())).to_db()
});
let found = scheduled_jobs::set_run_context(skald.db(), id, json.as_deref()).await?;
let found = scheduled_jobs::set_run_context(&ctx.pool, id, json.as_deref()).await?;
if found { Ok(()) } else { Err(ApiError::not_found(format!("job {id} not found"))) }
}
@@ -106,19 +116,26 @@ pub struct JobRunResponse {
}
pub async fn kill_job(
Path(id): Path<i64>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(id): Path<i64>,
) -> Result<StatusCode, ApiError> {
let job = scheduled_jobs::get_by_id(skald.db(), id).await?
let ctx = require_context(&skald, &auth.user_id).await?;
let job = scheduled_jobs::get_by_id(&ctx.pool, id).await?
.ok_or_else(|| ApiError::not_found(format!("job {id} not found")))?;
let session_id = job.running_session_id
.ok_or_else(|| ApiError::bad_request("job is not currently running"))?;
skald.system_bus().send(core_api::system_bus::SystemEvent::SessionCancelled { session_id });
// Direct cancel on the user's own session manager — no system bus fan-out.
ctx.sessions.cancel_session(session_id).await;
Ok(StatusCode::ACCEPTED)
}
pub async fn list_runs(State(skald): State<Arc<Skald>>) -> Result<Json<Vec<JobRunResponse>>, ApiError> {
let runs = job_runs::list_all(skald.db(), 200).await?;
pub async fn list_runs(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
) -> Result<Json<Vec<JobRunResponse>>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let runs = job_runs::list_all(&ctx.pool, 200).await?;
Ok(Json(runs.into_iter().map(|r| JobRunResponse {
id: r.id,
job_id: r.job_id,
src/frontend/api/dev.rs
+68 -63
View File
@@ -1,13 +1,13 @@
use axum::{
extract::{Path, Query, State},
response::IntoResponse,
Json,
Json, Extension,
};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use skald_core::skald::Skald;
use super::ApiError;
use super::{ApiError, guard::AuthUser, require_context};
const KEY: &str = "DEBUG_MODE";
@@ -44,8 +44,6 @@ const PAGE_SIZE: i64 = 20;
#[derive(Deserialize)]
pub struct LlmRequestsQuery {
pub agent_id: Option<String>,
pub source: Option<String>,
pub from: Option<String>,
pub to: Option<String>,
pub page: Option<i64>,
@@ -54,8 +52,6 @@ pub struct LlmRequestsQuery {
#[derive(Serialize)]
pub struct LlmRequestItem {
pub id: i64,
pub agent_id: Option<String>,
pub source: Option<String>,
pub model_name: String,
pub created_at: String,
pub input_tokens: Option<i64>,
@@ -76,37 +72,34 @@ pub struct LlmRequestsResponse {
pub async fn list_llm_requests(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Query(params): Query<LlmRequestsQuery>,
) -> Result<impl IntoResponse, ApiError> {
let page = params.page.unwrap_or(1).max(1);
let offset = (page - 1) * PAGE_SIZE;
// Bind optional filters twice each: once for the IS NULL check, once for the
// equality check. SQLite evaluates `? IS NULL` against the bound value itself.
let items = sqlx::query_as::<_, (i64, Option<String>, Option<String>, String, String, Option<i64>, Option<i64>, Option<i64>, Option<i64>, i64, Option<String>)>(
// Metadata-only query on system.db. No JOIN with chat_sessions (that table
// lives in the per-user owner bucket). Filters by user_id so each user sees
// only their own requests.
let items = sqlx::query_as::<_, (i64, String, String, Option<i64>, Option<i64>, Option<i64>, Option<i64>, i64, Option<String>)>(
"SELECT
r.id,
s.agent_id,
s.source,
r.model_name,
r.created_at,
r.input_tokens,
r.output_tokens,
r.cache_read_tokens,
r.cache_creation_tokens,
r.duration_ms,
r.error_text
FROM llm_requests r
LEFT JOIN chat_sessions s ON s.id = r.session_id
WHERE (? IS NULL OR s.agent_id = ?)
AND (? IS NULL OR s.source = ?)
AND (? IS NULL OR r.created_at >= ?)
AND (? IS NULL OR r.created_at <= ?)
ORDER BY r.created_at DESC
id,
model_name,
created_at,
input_tokens,
output_tokens,
cache_read_tokens,
cache_creation_tokens,
duration_ms,
error_text
FROM llm_requests
WHERE user_id = ?
AND (? IS NULL OR created_at >= ?)
AND (? IS NULL OR created_at <= ?)
ORDER BY created_at DESC
LIMIT ? OFFSET ?",
)
.bind(&params.agent_id).bind(&params.agent_id)
.bind(&params.source).bind(&params.source)
.bind(&auth.user_id)
.bind(&params.from).bind(&params.from)
.bind(&params.to).bind(&params.to)
.bind(PAGE_SIZE)
@@ -114,22 +107,19 @@ pub async fn list_llm_requests(
.fetch_all(&**skald.db())
.await?
.into_iter()
.map(|(id, agent_id, source, model_name, created_at, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, duration_ms, error_text)| {
LlmRequestItem { id, agent_id, source, model_name, created_at, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, duration_ms, error_text }
.map(|(id, model_name, created_at, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, duration_ms, error_text)| {
LlmRequestItem { id, model_name, created_at, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, duration_ms, error_text }
})
.collect::<Vec<_>>();
let total = sqlx::query_scalar::<_, i64>(
"SELECT COUNT(*)
FROM llm_requests r
LEFT JOIN chat_sessions s ON s.id = r.session_id
WHERE (? IS NULL OR s.agent_id = ?)
AND (? IS NULL OR s.source = ?)
AND (? IS NULL OR r.created_at >= ?)
AND (? IS NULL OR r.created_at <= ?)",
FROM llm_requests
WHERE user_id = ?
AND (? IS NULL OR created_at >= ?)
AND (? IS NULL OR created_at <= ?)",
)
.bind(&params.agent_id).bind(&params.agent_id)
.bind(&params.source).bind(&params.source)
.bind(&auth.user_id)
.bind(&params.from).bind(&params.from)
.bind(&params.to).bind(&params.to)
.fetch_one(&**skald.db())
@@ -143,8 +133,7 @@ pub async fn list_llm_requests(
#[derive(Serialize)]
pub struct LlmRequestDetail {
pub id: i64,
pub agent_id: Option<String>,
pub source: Option<String>,
pub request_id: Option<String>,
pub stack_id: Option<i64>,
pub model_name: String,
pub created_at: String,
@@ -154,6 +143,8 @@ pub struct LlmRequestDetail {
pub cache_creation_tokens: Option<i64>,
pub duration_ms: i64,
pub error_text: Option<String>,
// Payload fields — now live in llm_request_payloads in the user's own database.
// Left as None for now; a future cross-pool lookup via request_id can fill them.
pub request_json: Option<String>,
pub request_headers: Option<String>,
pub response_json: Option<String>,
@@ -162,40 +153,54 @@ pub struct LlmRequestDetail {
pub async fn get_llm_request(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(id): Path<i64>,
) -> Result<impl IntoResponse, ApiError> {
let row = sqlx::query_as::<_, (i64, Option<String>, Option<String>, Option<i64>, String, String, Option<i64>, Option<i64>, Option<i64>, Option<i64>, i64, Option<String>, Option<String>, Option<String>, Option<String>, Option<String>)>(
let row = sqlx::query_as::<_, (i64, Option<String>, Option<i64>, String, String, Option<i64>, Option<i64>, Option<i64>, Option<i64>, i64, Option<String>)>(
"SELECT
r.id,
s.agent_id,
s.source,
r.stack_id,
r.model_name,
r.created_at,
r.input_tokens,
r.output_tokens,
r.cache_read_tokens,
r.cache_creation_tokens,
r.duration_ms,
r.error_text,
NULLIF(r.request_json, '') AS request_json,
r.request_headers,
r.response_json,
r.response_headers
FROM llm_requests r
LEFT JOIN chat_sessions s ON s.id = r.session_id
WHERE r.id = ?",
id,
request_id,
stack_id,
model_name,
created_at,
input_tokens,
output_tokens,
cache_read_tokens,
cache_creation_tokens,
duration_ms,
error_text
FROM llm_requests
WHERE id = ? AND user_id = ?",
)
.bind(id)
.bind(&auth.user_id)
.fetch_optional(&**skald.db())
.await?;
let Some((id, agent_id, source, stack_id, model_name, created_at, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, duration_ms, error_text, request_json, request_headers, response_json, response_headers)) = row else {
let Some((id, request_id, stack_id, model_name, created_at, input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens, duration_ms, error_text)) = row else {
return Err(ApiError::not_found(format!("llm_request {id} not found")));
};
// Try to fetch the payload from the user's own database via request_id.
let payload = if let Some(ref rid) = request_id {
let ctx = require_context(&skald, &auth.user_id).await.ok();
if let Some(ctx) = ctx {
sqlx::query_as::<_, (Option<String>, Option<String>, Option<String>, Option<String>)>(
"SELECT request_json, request_headers, response_json, response_headers
FROM llm_request_payloads WHERE request_id = ?",
)
.bind(rid)
.fetch_optional(&*ctx.pool)
.await
.ok()
.flatten()
} else { None }
} else { None };
let (request_json, request_headers, response_json, response_headers) = payload.unwrap_or((None, None, None, None));
Ok(Json(LlmRequestDetail {
id, agent_id, source, stack_id, model_name, created_at,
id, request_id, stack_id, model_name, created_at,
input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens,
duration_ms, error_text,
request_json, request_headers, response_json, response_headers,
src/frontend/api/inbox.rs
+30 -20
View File
@@ -1,5 +1,5 @@
use axum::{
Json,
Json, Extension,
extract::{Path, State},
};
use serde::Deserialize;
@@ -7,22 +7,26 @@ use serde_json::{Value, json};
use std::time::Duration;
use std::sync::Arc;
use skald_core::skald::Skald;
use super::ApiError;
use skald_core::skald::{Skald, UserContext};
use super::{ApiError, guard::AuthUser, require_context};
// ── GET /api/inbox ────────────────────────────────────────────────────────────
//
// Returns all pending approval requests and clarification requests in a single
// response, so the frontend can show a unified Agent Inbox page with one fetch.
pub async fn list(State(skald): State<Arc<Skald>>) -> Json<Value> {
let items = skald.inbox().list_pending().await;
Json(json!({
pub async fn list(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
) -> Result<Json<Value>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let items = ctx.inbox.list_pending().await;
Ok(Json(json!({
"total": items.total,
"approvals": items.approvals,
"clarifications": items.clarifications,
"elicitations": items.elicitations,
}))
})))
}
// ── POST /api/inbox/approvals/:request_id/resolve ─────────────────────────────
@@ -47,17 +51,19 @@ fn default_action() -> String { "approve".to_string() }
pub async fn resolve_approval(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ApprovePath>,
Json(body): Json<ApproveBody>,
) -> Result<Json<Value>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
// Peek info before resolving so we have session_id and tool metadata for bypass.
let info = skald.approval().get_pending(p.request_id).await;
let info = ctx.approval.get_pending(p.request_id).await;
if body.action == "reject" {
// Pass the raw note; the waiting session builds the canonical message.
skald.inbox().reject(p.request_id, body.note.clone()).await;
ctx.inbox.reject(p.request_id, body.note.clone()).await;
} else {
skald.inbox().approve(p.request_id).await;
ctx.inbox.approve(p.request_id).await;
// Apply bypass if requested (only on approve).
if let (Some(info), Some(bypass_secs)) = (info, body.bypass_secs) {
@@ -72,29 +78,29 @@ pub async fn resolve_approval(
match scope {
"category" => {
if let Some(cat) = info.tool_category {
skald.approval().bypass_session_for_category(info.session_id, cat, duration).await;
ctx.approval.bypass_session_for_category(info.session_id, cat, duration).await;
} else {
apply_all_bypass(&skald, info.session_id, duration).await;
apply_all_bypass(&ctx, info.session_id, duration).await;
}
}
"mcp_server" => {
if let Some(server) = info.mcp_server {
skald.approval().bypass_session_for_mcp(info.session_id, server, duration).await;
ctx.approval.bypass_session_for_mcp(info.session_id, server, duration).await;
} else {
apply_all_bypass(&skald, info.session_id, duration).await;
apply_all_bypass(&ctx, info.session_id, duration).await;
}
}
_ => apply_all_bypass(&skald, info.session_id, duration).await,
_ => apply_all_bypass(&ctx, info.session_id, duration).await,
}
}
}
Ok(Json(json!({ "ok": true, "request_id": p.request_id, "action": body.action })))
}
async fn apply_all_bypass(skald: &Skald, session_id: i64, duration: Option<Duration>) {
async fn apply_all_bypass(ctx: &UserContext, session_id: i64, duration: Option<Duration>) {
match duration {
Some(d) => skald.approval().bypass_session_for(session_id, d).await,
None => skald.approval().bypass_session(session_id).await,
Some(d) => ctx.approval.bypass_session_for(session_id, d).await,
None => ctx.approval.bypass_session(session_id).await,
}
}
@@ -110,13 +116,15 @@ pub struct ClarifyBody {
pub async fn resolve_clarification(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ClarifyPath>,
Json(body): Json<ClarifyBody>,
) -> Result<Json<Value>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
if body.answer.trim().is_empty() {
return Err(ApiError::bad_request("answer must not be empty"));
}
let resolved = skald.inbox().answer(p.request_id, body.answer).await;
let resolved = ctx.inbox.answer(p.request_id, body.answer).await;
if resolved {
Ok(Json(json!({ "ok": true, "request_id": p.request_id })))
} else {
@@ -145,14 +153,16 @@ fn default_elicit_action() -> String { "decline".to_string() }
pub async fn resolve_elicitation(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ElicitPath>,
Json(body): Json<ElicitBody>,
) -> Result<Json<Value>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let action = match body.action.as_str() {
"accept" | "decline" | "cancel" => body.action.clone(),
other => return Err(ApiError::bad_request(format!("invalid action: {other}"))),
};
let resolved = skald.inbox().resolve_elicitation(p.request_id, action.clone(), body.content).await;
let resolved = ctx.inbox.resolve_elicitation(p.request_id, action.clone(), body.content).await;
if resolved {
Ok(Json(json!({ "ok": true, "request_id": p.request_id, "action": action })))
} else {
src/frontend/api/llm.rs
+13 -13
View File
@@ -17,7 +17,7 @@ pub async fn provider_models(
State(skald): State<Arc<Skald>>,
axum::extract::Path(id): axum::extract::Path<i64>,
) -> Result<Json<Vec<RemoteLlmModelInfo>>, ApiError> {
let models = skald.manager().llm_manager().list_provider_models(id).await?;
let models = skald.llm_manager().list_provider_models(id).await?;
Ok(Json(models))
}
@@ -36,7 +36,7 @@ pub async fn provider_reasoning_mode(
axum::extract::Path(id): axum::extract::Path<i64>,
axum::extract::Query(q): axum::extract::Query<ReasoningModeQuery>,
) -> Json<Option<ReasoningMode>> {
let mode = skald.manager().llm_manager().reasoning_mode_for(id, &q.model_id).await;
let mode = skald.llm_manager().reasoning_mode_for(id, &q.model_id).await;
Json(mode)
}
@@ -51,7 +51,7 @@ pub struct SelectorResponse {
pub async fn selector(
State(skald): State<Arc<Skald>>,
) -> Result<Json<SelectorResponse>, ApiError> {
let mgr = skald.manager().llm_manager();
let mgr = skald.llm_manager();
let models = mgr.client_names().await;
let default = mgr.default_name().await;
Ok(Json(SelectorResponse { models, default }))
@@ -62,7 +62,7 @@ pub async fn selector(
pub async fn list_providers(
State(skald): State<Arc<Skald>>,
) -> Result<Json<Vec<LlmProviderInfo>>, ApiError> {
Ok(Json(skald.manager().llm_manager().list_providers_info().await))
Ok(Json(skald.llm_manager().list_providers_info().await))
}
#[derive(Deserialize)]
@@ -94,7 +94,7 @@ pub async fn create_provider(
) -> Result<StatusCode, ApiError> {
validate_provider_type(&skald, &payload.provider)?;
let record = LlmProviderRecord::from(payload);
skald.manager().llm_manager().add_provider(record).await?;
skald.llm_manager().add_provider(record).await?;
Ok(StatusCode::CREATED)
}
@@ -102,7 +102,7 @@ pub async fn get_provider(
State(skald): State<Arc<Skald>>,
axum::extract::Path(id): axum::extract::Path<i64>,
) -> Result<Json<LlmProviderRecord>, ApiError> {
skald.manager().llm_manager().get_provider(id).await
skald.llm_manager().get_provider(id).await
.map(Json)
.ok_or_else(|| ApiError::not_found(format!("provider {id} not found")))
}
@@ -114,7 +114,7 @@ pub async fn update_provider(
) -> Result<StatusCode, ApiError> {
validate_provider_type(&skald, &payload.provider)?;
let record = LlmProviderRecord::from(payload);
skald.manager().llm_manager().update_provider(id, record).await?;
skald.llm_manager().update_provider(id, record).await?;
Ok(StatusCode::NO_CONTENT)
}
@@ -122,7 +122,7 @@ pub async fn delete_provider(
State(skald): State<Arc<Skald>>,
axum::extract::Path(id): axum::extract::Path<i64>,
) -> Result<StatusCode, ApiError> {
skald.manager().llm_manager().delete_provider(id).await?;
skald.llm_manager().delete_provider(id).await?;
Ok(StatusCode::NO_CONTENT)
}
@@ -131,7 +131,7 @@ pub async fn delete_provider(
pub async fn list_models(
State(skald): State<Arc<Skald>>,
) -> Result<Json<Vec<LlmModelInfo>>, ApiError> {
let mgr = skald.manager().llm_manager();
let mgr = skald.llm_manager();
// Warm the catalog cache for every provider concurrently so that price data
// is available for the join inside list_models_info(). Errors are ignored —
@@ -203,7 +203,7 @@ pub async fn create_model(
Json(payload): Json<ModelPayload>,
) -> Result<StatusCode, ApiError> {
let record = LlmModelRecord::try_from(payload)?;
skald.manager().llm_manager().add_model(record).await?;
skald.llm_manager().add_model(record).await?;
Ok(StatusCode::CREATED)
}
@@ -211,7 +211,7 @@ pub async fn get_model(
State(skald): State<Arc<Skald>>,
axum::extract::Path(id): axum::extract::Path<i64>,
) -> Result<Json<LlmModelRecord>, ApiError> {
skald.manager().llm_manager().get_model(id).await
skald.llm_manager().get_model(id).await
.map(Json)
.ok_or_else(|| ApiError::not_found(format!("model {id} not found")))
}
@@ -222,7 +222,7 @@ pub async fn update_model(
Json(payload): Json<ModelPayload>,
) -> Result<StatusCode, ApiError> {
let record = LlmModelRecord::try_from(payload)?;
skald.manager().llm_manager().update_model(id, record).await?;
skald.llm_manager().update_model(id, record).await?;
Ok(StatusCode::NO_CONTENT)
}
@@ -230,7 +230,7 @@ pub async fn delete_model(
State(skald): State<Arc<Skald>>,
axum::extract::Path(id): axum::extract::Path<i64>,
) -> Result<StatusCode, ApiError> {
skald.manager().llm_manager().delete_model(id).await?;
skald.llm_manager().delete_model(id).await?;
Ok(StatusCode::NO_CONTENT)
}
src/frontend/api/mod.rs
+16 -9
View File
@@ -139,8 +139,6 @@ pub fn router() -> Router<Arc<Skald>> {
// Config properties
.route("/config", get(config::list_properties))
.route("/config/{key}", put(config::set_property))
// TIC
.route("/tic/trigger", post(tic_trigger))
// Plugins
.route("/plugins", get(plugins::list))
.route("/plugins/{id}", put(plugins::update))
@@ -164,13 +162,6 @@ pub fn router() -> Router<Arc<Skald>> {
.route("/file", delete(files::delete_file))
}
async fn tic_trigger(State(skald): State<Arc<Skald>>) -> impl IntoResponse {
tokio::spawn(async move {
Arc::clone(skald.tic_manager()).tick_now().await;
});
StatusCode::ACCEPTED
}
pub struct ApiError {
status: StatusCode,
message: String,
@@ -184,6 +175,22 @@ impl ApiError {
pub fn not_found(msg: impl Into<String>) -> Self {
Self { status: StatusCode::NOT_FOUND, message: msg.into() }
}
pub fn unauthorized(msg: impl Into<String>) -> Self {
Self { status: StatusCode::UNAUTHORIZED, message: msg.into() }
}
}
/// Resolves the authenticated caller's per-user runtime context, or `401` when the
/// user's database is locked (not logged in). Every owner-bound HTTP handler starts
/// here to route reads/writes to the caller's own `{userid}.db` instead of the
/// shared registry pool. The `AuthUser` is injected by [`guard::require_auth`].
pub async fn require_context(
skald: &Skald,
user_id: &str,
) -> Result<Arc<skald_core::skald::UserContext>, ApiError> {
skald.user_context(user_id).await
.ok_or_else(|| ApiError::unauthorized("session expired — please log in again"))
}
impl IntoResponse for ApiError {
src/frontend/api/projects.rs
+93 -40
View File
@@ -1,17 +1,19 @@
use std::sync::Arc;
use axum::{
Json,
Json, Extension,
extract::{Path, State},
http::StatusCode,
};
use serde::{Deserialize, Serialize};
use sqlx::SqlitePool;
use skald_core::db::project_tickets::ProjectTicket;
use skald_core::db::projects::Project;
use skald_core::db::{project_tickets, projects};
use skald_core::run_context::RunContext;
use skald_core::skald::Skald;
use super::ApiError;
use super::{ApiError, guard::AuthUser, require_context};
/// Source-id prefix for a project's interactive chat session (e.g. `project-42`).
/// A hyphen (not `:`) is used so the id is URL-safe in `/api/{source}/messages`.
@@ -130,62 +132,70 @@ impl<'de> Deserialize<'de> for TicketPath {
pub async fn list(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
) -> Result<Json<Vec<ProjectResponse>>, ApiError> {
let projects = skald.projects().list().await?;
Ok(Json(projects.into_iter().map(Into::into).collect()))
let ctx = require_context(&skald, &auth.user_id).await?;
let items = projects::list(&ctx.pool).await?;
Ok(Json(items.into_iter().map(Into::into).collect()))
}
pub async fn create(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Json(body): Json<ProjectBody>,
) -> Result<(StatusCode, Json<ProjectResponse>), ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let rc_json = body.rc_json();
let rc = rc_json.as_deref().and_then(RunContext::from_db);
let project = skald.projects().create(
let project = projects::create(
&ctx.pool,
&body.name,
&body.path,
body.description.as_deref().unwrap_or(""),
rc.as_ref(),
rc_json.as_deref(),
).await?;
Ok((StatusCode::CREATED, Json(project.into())))
}
pub async fn get_project(
Path(p): Path<ProjectPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ProjectPath>,
) -> Result<Json<ProjectResponse>, ApiError> {
let project = skald.projects().get(p.id).await?
let ctx = require_context(&skald, &auth.user_id).await?;
let project = projects::get(&ctx.pool, p.id).await?
.ok_or_else(|| ApiError::not_found(format!("project {} not found", p.id)))?;
Ok(Json(project.into()))
}
pub async fn update(
Path(p): Path<ProjectPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ProjectPath>,
Json(body): Json<ProjectBody>,
) -> Result<Json<ProjectResponse>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let rc_json = body.rc_json();
let rc = rc_json.as_deref().and_then(RunContext::from_db);
let found = skald.projects().update(
p.id,
&body.name,
&body.path,
let found = projects::update(
&ctx.pool, p.id,
&body.name, &body.path,
body.description.as_deref().unwrap_or(""),
rc.as_ref(),
rc_json.as_deref(),
).await?;
if !found {
return Err(ApiError::not_found(format!("project {} not found", p.id)));
}
let project = skald.projects().get(p.id).await?
let project = projects::get(&ctx.pool, p.id).await?
.ok_or_else(|| ApiError::not_found(format!("project {} not found", p.id)))?;
Ok(Json(project.into()))
}
pub async fn delete(
Path(p): Path<ProjectPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ProjectPath>,
) -> Result<StatusCode, ApiError> {
let found = skald.projects().delete(p.id).await?;
let ctx = require_context(&skald, &auth.user_id).await?;
let found = projects::delete(&ctx.pool, p.id).await?;
if found { Ok(StatusCode::NO_CONTENT) }
else { Err(ApiError::not_found(format!("project {} not found", p.id))) }
}
@@ -193,56 +203,97 @@ pub async fn delete(
// ── Ticket handlers ───────────────────────────────────────────────────────────
pub async fn list_tickets(
Path(p): Path<ProjectPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ProjectPath>,
) -> Result<Json<Vec<TicketResponse>>, ApiError> {
let tickets = skald.ticket_manager().list(p.id).await?;
let ctx = require_context(&skald, &auth.user_id).await?;
let tickets = project_tickets::list_for_project(&ctx.pool, p.id).await?;
Ok(Json(tickets.into_iter().map(Into::into).collect()))
}
pub async fn create_ticket(
Path(p): Path<ProjectPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ProjectPath>,
Json(body): Json<TicketBody>,
) -> Result<(StatusCode, Json<TicketResponse>), ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let rc_json = body.rc_json();
let rc = rc_json.as_deref().and_then(RunContext::from_db);
// Tickets run a task sub-agent — no default. The agent's `type == task` is enforced
// when the ticket starts, via TaskManager::spawn_async_job (require_task_agent).
let agent_id = body.agent_id.as_deref().map(str::trim).filter(|s| !s.is_empty())
.ok_or_else(|| ApiError::bad_request("agent_id is required — pick a task agent for this ticket"))?;
let ticket = skald.ticket_manager().create(
p.id,
let ticket = project_tickets::create(
&ctx.pool, p.id,
&body.title,
body.description.as_deref().unwrap_or(""),
agent_id,
rc.as_ref(),
rc_json.as_deref(),
).await?;
projects::touch(&ctx.pool, p.id).await?;
Ok((StatusCode::CREATED, Json(ticket.into())))
}
pub async fn delete_ticket(
Path(tp): Path<TicketPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(tp): Path<TicketPath>,
) -> Result<StatusCode, ApiError> {
let found = skald.ticket_manager().delete(tp.tid).await?;
if found { Ok(StatusCode::NO_CONTENT) }
else { Err(ApiError::not_found(format!("ticket {} not found", tp.tid))) }
let ctx = require_context(&skald, &auth.user_id).await?;
let ticket = project_tickets::get(&ctx.pool, tp.tid).await?;
let found = project_tickets::delete(&ctx.pool, tp.tid).await?;
if found {
if let Some(t) = ticket {
projects::touch(&ctx.pool, t.project_id).await?;
}
Ok(StatusCode::NO_CONTENT)
} else {
Err(ApiError::not_found(format!("ticket {} not found", tp.tid)))
}
}
pub async fn start_ticket(
Path(tp): Path<TicketPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(tp): Path<TicketPath>,
) -> Result<StatusCode, ApiError> {
skald.ticket_manager().start(tp.tid).await?;
let ctx = require_context(&skald, &auth.user_id).await?;
let ticket = project_tickets::get(&ctx.pool, tp.tid).await?
.ok_or_else(|| ApiError::not_found(format!("ticket {} not found", tp.tid)))?;
let project = projects::get(&ctx.pool, ticket.project_id).await?
.ok_or_else(|| ApiError::not_found(format!("project {} not found", ticket.project_id)))?;
let base: Option<RunContext> =
ticket.run_context.as_deref().and_then(RunContext::from_db)
.or_else(|| project.run_context.as_deref().and_then(RunContext::from_db));
let rc = skald_core::projects::build_runtime_run_context(&project, base);
let origin_ref = format!("PROJECT_TASK:{}", tp.tid);
let rc_json = rc.to_db();
let job = ctx.cron.spawn_async_job(
&ticket.title,
&ticket.description,
&ticket.description,
&ticket.agent_id,
Some(&rc_json),
&origin_ref,
)?;
project_tickets::start(&ctx.pool, tp.tid, job.id).await?;
projects::touch(&ctx.pool, ticket.project_id).await?;
Ok(StatusCode::ACCEPTED)
}
pub async fn reset_ticket(
Path(tp): Path<TicketPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(tp): Path<TicketPath>,
) -> Result<StatusCode, ApiError> {
skald.ticket_manager().reset(tp.tid).await?;
let ctx = require_context(&skald, &auth.user_id).await?;
let project_id = project_tickets::get(&ctx.pool, tp.tid).await?.map(|t| t.project_id);
project_tickets::reset(&ctx.pool, tp.tid).await?;
if let Some(pid) = project_id {
projects::touch(&ctx.pool, pid).await?;
}
Ok(StatusCode::NO_CONTENT)
}
@@ -261,7 +312,7 @@ pub struct SessionResponse {
/// provisioning config, shared by session-open and session-reset so the two never
/// diverge.
pub async fn provisioning_for_source(
skald: &Skald,
pool: &SqlitePool,
source: &str,
) -> Result<(String, Option<RunContext>), ApiError> {
let Some(id) = source
@@ -271,7 +322,7 @@ pub async fn provisioning_for_source(
return Ok(("main".to_string(), None));
};
let project = skald.projects().get(id).await?
let project = projects::get(pool, id).await?
.ok_or_else(|| ApiError::not_found(format!("project {id} not found")))?;
let base = project.run_context.as_deref().and_then(RunContext::from_db);
let rc = skald_core::projects::build_runtime_run_context(&project, base);
@@ -284,10 +335,12 @@ pub async fn provisioning_for_source(
pub async fn open_session(
Path(p): Path<ProjectPath>,
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
) -> Result<Json<SessionResponse>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let source = format!("{PROJECT_SOURCE_PREFIX}{}", p.id);
let (agent, rc) = provisioning_for_source(&skald, &source).await?;
let session_id = skald.chat_hub()
let (agent, rc) = provisioning_for_source(&ctx.pool, &source).await?;
let session_id = ctx.chat_hub
.provision_session(&source, &agent, rc.as_ref(), false)
.await?;
Ok(Json(SessionResponse { source, session_id }))
src/frontend/api/run_context.rs
+13 -4
View File
@@ -1,7 +1,7 @@
use std::sync::Arc;
use axum::{
Json,
Json, Extension,
extract::{Path, State},
http::StatusCode,
};
@@ -9,7 +9,7 @@ use serde::Deserialize;
use serde_json::{Value, json};
use skald_core::skald::Skald;
use super::ApiError;
use super::{ApiError, guard::AuthUser, require_context};
// ── Tool Permission Groups ────────────────────────────────────────────────────
@@ -87,12 +87,21 @@ pub struct SessionPath { pub session_id: i64 }
/// POST body: the full RunContext object, or JSON `null` to clear the context.
pub async fn set_session_run_context(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<SessionPath>,
Json(ctx): Json<Option<skald_core::run_context::RunContext>>,
) -> Result<Json<Value>, ApiError> {
skald.run_context_manager().set_session_run_context(p.session_id, ctx.as_ref()).await?;
let uctx = require_context(&skald, &auth.user_id).await?;
// The session row (and its live handler) live in the caller's own pool, so the
// persist + live update both target the user's context. Run-context *definitions*
// (roles) remain instance-wide; only the per-session value is owner data.
skald_core::db::chat_sessions::set_run_context(
&uctx.pool,
p.session_id,
ctx.as_ref().map(|c| c.to_db()).as_deref(),
).await?;
if let Some(handler) = skald.manager().active_handler(p.session_id).await {
if let Some(handler) = uctx.sessions.active_handler(p.session_id).await {
handler.set_run_context(ctx).await;
}
src/frontend/api/sessions.rs
+46 -28
View File
@@ -3,7 +3,7 @@ use std::future::Future;
use std::pin::Pin;
use axum::{
Json,
Json, Extension,
extract::{Path, Query, State},
};
use serde::{Deserialize, Serialize};
@@ -13,12 +13,12 @@ use sqlx::SqlitePool;
use skald_core::db::{chat_history, chat_llm_tools, chat_sessions, chat_sessions_stack, sources};
use skald_core::db::chat_sessions_stack::SessionStack;
use std::sync::Arc;
use skald_core::skald::Skald;
use skald_core::skald::{Skald, UserContext};
use skald_core::session::handler::ApprovalDecision;
use skald_core::approval::ApprovalManager;
use skald_core::tools::{ToolRegistry, ToolDescriptionLength, tool_names as tn};
use super::ApiError;
use super::{ApiError, guard::AuthUser, require_context};
// ── POST /api/sessions — start a new conversation ─────────────────────────────
@@ -32,12 +32,14 @@ fn default_source() -> String { "web".to_string() }
pub async fn create(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Query(q): Query<CreateQuery>,
) -> Result<Json<Value>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
// Resolve agent + RunContext from the source so project chats reset with the
// coordinator agent (not the default `main`), then provision a fresh session.
let (agent, rc) = super::projects::provisioning_for_source(&skald, &q.source).await?;
skald.chat_hub().provision_session(&q.source, &agent, rc.as_ref(), true).await?;
let (agent, rc) = super::projects::provisioning_for_source(&ctx.pool, &q.source).await?;
ctx.chat_hub.provision_session(&q.source, &agent, rc.as_ref(), true).await?;
Ok(Json(json!({})))
}
@@ -45,8 +47,10 @@ pub async fn create(
pub async fn web_messages(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
) -> Result<Json<Vec<Value>>, ApiError> {
messages_for_source(&skald, "web").await
let ctx = require_context(&skald, &auth.user_id).await?;
messages_for_source(&skald, &ctx, "web").await
}
// ── GET /api/:source/messages ─────────────────────────────────────────────────
@@ -56,31 +60,36 @@ pub struct SourcePath { pub source: String }
pub async fn source_messages(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<SourcePath>,
) -> Result<Json<Vec<Value>>, ApiError> {
messages_for_source(&skald, &p.source).await
let ctx = require_context(&skald, &auth.user_id).await?;
messages_for_source(&skald, &ctx, &p.source).await
}
async fn messages_for_source(skald: &Arc<Skald>, source: &str) -> Result<Json<Vec<Value>>, ApiError> {
let session_id = match sources::active_session_id(skald.db(), source).await? {
async fn messages_for_source(skald: &Arc<Skald>, ctx: &UserContext, source: &str) -> Result<Json<Vec<Value>>, ApiError> {
// History/sessions read from the caller's own pool; the tool registry is a
// global capability, and approval is this user's per-user manager.
let db = &ctx.pool;
let session_id = match sources::active_session_id(db, source).await? {
Some(id) => id,
None => return Ok(Json(vec![])),
};
let main_stack = match chat_sessions_stack::main_for_session(skald.db(), session_id).await? {
let main_stack = match chat_sessions_stack::main_for_session(db, session_id).await? {
Some(s) => s,
None => return Ok(Json(vec![])),
};
let subagent_map: HashMap<i64, SessionStack> =
chat_sessions_stack::all_for_session(skald.db(), session_id)
chat_sessions_stack::all_for_session(db, session_id)
.await?
.into_iter()
.filter_map(|s| s.parent_tool_call_id.map(|tc_id| (tc_id, s)))
.collect();
let mut items: Vec<Value> = Vec::new();
build_items(skald.db(), skald.tools(), skald.approval(), &main_stack, &subagent_map, &mut items).await?;
build_items(db, skald.tools(), &ctx.approval, &main_stack, &subagent_map, &mut items).await?;
Ok(Json(items))
}
@@ -117,9 +126,12 @@ pub struct ResolveToolResponse {
/// resolve against the correct session — there is no "current session" scoping.
pub async fn resolve_tool(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<ResolveToolPath>,
Json(body): Json<ResolveToolBody>,
) -> Result<Json<ResolveToolResponse>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let db = &ctx.pool;
// Look up the tool call by id alone — no active-session filter. Also pull the
// owning session_id so the post-restart path drives the correct session.
let tc = sqlx::query_as::<_, (i64, String, Option<String>, String, i64)>(
@@ -130,7 +142,7 @@ pub async fn resolve_tool(
WHERE t.id = ?",
)
.bind(p.tool_call_id)
.fetch_optional(&**skald.db())
.fetch_optional(&**db)
.await?
.ok_or_else(|| anyhow::anyhow!(
"tool_call_id {} not found", p.tool_call_id
@@ -152,12 +164,12 @@ pub async fn resolve_tool(
// Pass the raw user note to the live session so the loop builds the
// canonical message; for the not-live path (no waiting session, e.g.
// after a restart) build the same message here and save it directly.
let live = skald.approval()
let live = ctx.approval
.resolve_for_tool_call(tc_id, ApprovalDecision::Rejected { note: body.note.clone() })
.await;
let msg = ApprovalDecision::rejection_message(&body.note);
if !live {
chat_llm_tools::reject(skald.db(), tc_id, &msg).await?;
chat_llm_tools::reject(db, tc_id, &msg).await?;
}
return Ok(Json(ResolveToolResponse {
tool_call_id: tc_id,
@@ -169,7 +181,7 @@ pub async fn resolve_tool(
// `restart` calls process::exit — mark done in DB first.
if tc_name == tn::RESTART {
chat_llm_tools::complete(skald.db(), tc_id, "Riavvio avviato.", "string").await?;
chat_llm_tools::complete(db, tc_id, "Riavvio avviato.", "string").await?;
// Use _exit() to skip C atexit handlers (e.g. Metal GPU cleanup in
// whisper-rs/ggml, which aborts with SIGABRT and yields exit code 134
// instead of 255 — breaking the run.sh restart supervisor).
@@ -177,7 +189,7 @@ pub async fn resolve_tool(
}
// ── Live path: LLM loop is blocked waiting for approval ──────────────────
if skald.approval()
if ctx.approval
.resolve_for_tool_call(tc_id, ApprovalDecision::Approved)
.await
{
@@ -196,9 +208,9 @@ pub async fn resolve_tool(
// via `execute_tool_call` (gate skipped) and continues the loop. Events stream
// to the reconnected client through the global bus; return immediately.
if tc_name == "execute_task" || tc_name == tn::EXECUTE_SUBTASK || tc_name == "run_subtask" {
let handler = skald.chat_hub().handler_for_session(session_id).await?;
let handler = ctx.chat_hub.handler_for_session(session_id).await?;
handler.mark_pre_approved(tc_id);
let hub = skald.chat_hub().clone();
let hub = ctx.chat_hub.clone();
tokio::spawn(async move {
if let Err(e) = hub.resume_session(session_id).await {
tracing::warn!(session_id, tool_call_id = tc_id, error = %e, "post-restart resume of sub-agent tool failed");
@@ -213,12 +225,12 @@ pub async fn resolve_tool(
}
// Simple tools: execute directly on the owning session and return the result.
let handler = skald.chat_hub().handler_for_session(session_id).await?;
let handler = ctx.chat_hub.handler_for_session(session_id).await?;
match handler.execute_tool(&tc_name, args).await {
Ok(result) => {
let wire = result.to_wire();
let kind = result.kind();
chat_llm_tools::complete(skald.db(), tc_id, &wire, kind).await?;
chat_llm_tools::complete(db, tc_id, &wire, kind).await?;
Ok(Json(ResolveToolResponse {
tool_call_id: tc_id,
status: "done".to_string(),
@@ -228,7 +240,7 @@ pub async fn resolve_tool(
}
Err(e) => {
let msg = e.to_string();
chat_llm_tools::fail(skald.db(), tc_id, &msg).await?;
chat_llm_tools::fail(db, tc_id, &msg).await?;
Err(anyhow::anyhow!(msg).into())
}
}
@@ -250,8 +262,11 @@ fn default_per_page() -> i64 { 20 }
pub async fn list_sessions(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Query(q): Query<ListSessionsQuery>,
) -> Result<Json<Value>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
let db = &ctx.pool;
let per_page = q.per_page.max(1).min(100);
let offset = ((q.page.max(1)) - 1) * per_page;
let src = q.source.as_deref();
@@ -261,7 +276,7 @@ pub async fn list_sessions(
WHERE (? IS NULL OR cs.source = ?)",
)
.bind(src).bind(src)
.fetch_one(&**skald.db()).await?;
.fetch_one(&**db).await?;
let rows = sqlx::query_as::<_, (i64, String, String, bool, bool, Option<String>, i64, Option<String>)>(
"SELECT cs.id, cs.source, cs.agent_id, cs.is_ephemeral, cs.is_interactive,
@@ -278,7 +293,7 @@ pub async fn list_sessions(
)
.bind(src).bind(src)
.bind(per_page).bind(offset)
.fetch_all(&**skald.db()).await?;
.fetch_all(&**db).await?;
let items: Vec<Value> = rows.into_iter().map(|(id, source, agent_id, is_ephemeral, is_interactive, created_at, message_count, last_message_at)| {
json!({
@@ -308,9 +323,12 @@ pub struct SessionIdPath { pub id: i64 }
pub async fn get_session_detail(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<SessionIdPath>,
) -> Result<Json<Value>, ApiError> {
let session = chat_sessions::find_by_id(skald.db(), p.id)
let ctx = require_context(&skald, &auth.user_id).await?;
let db = &ctx.pool;
let session = chat_sessions::find_by_id(db, p.id)
.await?
.ok_or_else(|| ApiError::not_found(format!("session {} not found", p.id)))?;
@@ -318,10 +336,10 @@ pub async fn get_session_detail(
"SELECT created_at FROM chat_sessions WHERE id = ?",
)
.bind(p.id)
.fetch_optional(&**skald.db())
.fetch_optional(&**db)
.await?;
let all_stacks = chat_sessions_stack::all_for_session(skald.db(), session.id).await?;
let all_stacks = chat_sessions_stack::all_for_session(db, session.id).await?;
let subagent_map: HashMap<i64, SessionStack> = all_stacks
.iter()
@@ -340,7 +358,7 @@ pub async fn get_session_detail(
};
let mut messages: Vec<Value> = Vec::new();
build_debug_items(skald.db(), skald.tools(), &main_stack, &subagent_map, &mut messages).await?;
build_debug_items(db, skald.tools(), &main_stack, &subagent_map, &mut messages).await?;
Ok(Json(json!({
"session": {
src/frontend/api/uploads.rs
+5 -3
View File
@@ -2,7 +2,7 @@ use std::path::{Path as StdPath, PathBuf};
use std::sync::Arc;
use axum::{
Json,
Json, Extension,
extract::{Multipart, Path, State},
};
use tokio::io::AsyncWriteExt;
@@ -11,7 +11,7 @@ use core_api::message_meta::Attachment;
use skald_core::skald::Skald;
use skald_core::tools::fs as fs_tools;
use super::ApiError;
use super::{ApiError, guard::AuthUser, require_context};
use super::sessions::SourcePath;
/// `POST /api/{source}/uploads`
@@ -25,12 +25,14 @@ use super::sessions::SourcePath;
/// size) so the client can show chips and echo them back when sending the message.
pub async fn upload(
State(skald): State<Arc<Skald>>,
Extension(auth): Extension<AuthUser>,
Path(p): Path<SourcePath>,
mut multipart: Multipart,
) -> Result<Json<Vec<Attachment>>, ApiError> {
let ctx = require_context(&skald, &auth.user_id).await?;
// Resolve (creating if needed) the source's session so uploads land in the
// directory the message will reference.
let session_id = skald.chat_hub().session_handler(&p.source).await?.session_id;
let session_id = ctx.chat_hub.session_handler(&p.source).await?.session_id;
let dir_rel = format!("data/uploads/{session_id}");
let dir_abs = fs_tools::resolve(&dir_rel)?;
src/frontend/api/ws.rs
+39 -19
View File
@@ -6,17 +6,20 @@ use axum::{
ws::{Message, WebSocket, WebSocketUpgrade},
},
response::IntoResponse,
Extension,
};
use serde::Deserialize;
use serde_json::Value;
use tokio::sync::broadcast;
use tracing::{debug, info, warn};
use skald_core::chat_hub::{ModelCommandOutcome, SendMessageOptions};
use skald_core::chat_hub::{ChatHub, ModelCommandOutcome, SendMessageOptions};
use skald_core::events::{ClientMessage, ServerEvent};
use skald_core::skald::Skald;
use core_api::command::CommandApi;
use super::guard::AuthUser;
#[derive(Deserialize)]
pub struct WsParams {
source: Option<String>,
@@ -68,16 +71,33 @@ fn dynamic_help(skald: &Skald) -> String {
pub async fn handler(
ws: WebSocketUpgrade,
Query(params): Query<WsParams>,
Extension(auth): Extension<AuthUser>,
State(skald): State<Arc<Skald>>,
) -> impl IntoResponse {
let source = params.source.unwrap_or_else(|| "web".to_string());
ws.on_upgrade(move |socket| handle_socket(socket, skald, source))
ws.on_upgrade(move |socket| handle_socket(socket, skald, source, auth.user_id))
}
// ── Socket loop ───────────────────────────────────────────────────────────────
async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String) {
let session_handler = match skald.chat_hub().session_handler(&source).await {
async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String, user_id: String) {
// Resolve the caller's per-user runtime. The pool is unlocked at login, so an
// authenticated connection normally has a context; a missing one means the
// database re-locked (e.g. a restart with no re-login) — report and close.
let ctx = match skald.user_context(&user_id).await {
Some(c) => c,
None => {
let _ = socket.send(to_msg(&ServerEvent::Error {
message: "session expired — please log in again".to_string(),
})).await;
return;
}
};
// Every chat operation for this connection goes through the user's own hub, so
// sessions land in their `{userid}.db` and events never cross to another user.
let chat_hub: Arc<ChatHub> = Arc::clone(&ctx.chat_hub);
let session_handler = match chat_hub.session_handler(&source).await {
Ok(h) => h,
Err(e) => {
let _ = socket.send(to_msg(&ServerEvent::Error { message: e.to_string() })).await;
@@ -85,9 +105,9 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
}
};
info!(source, "WebSocket connected");
info!(source, user = %user_id, "WebSocket connected");
let mut rx = skald.chat_hub().events(&source);
let mut rx = chat_hub.events(&source);
// Tell this (possibly reloaded) client whether a turn is already running for
// its session, so it can restore the STOP button. Sent after subscribing to
@@ -109,7 +129,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
// ── resume ────────────────────────────────────────────────────
if is_resume_msg(&text) {
info!("web WS: resume requested");
let hub = Arc::clone(skald.chat_hub());
let hub = Arc::clone(&chat_hub);
let src = source.clone();
tokio::spawn(async move {
if let Err(e) = hub.resume(&src).await {
@@ -127,10 +147,10 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
session_handler.cancel_pending_questions().await;
continue;
}
if handle_approval_msg(&text, skald.chat_hub()).await { continue; }
if handle_approval_msg(&text, &chat_hub).await { continue; }
if handle_question_answer_msg(&text, &session_handler).await { continue; }
if handle_data_msg(&text, &skald) { continue; }
if handle_select_client_msg(&text, &source, skald.chat_hub()).await { continue; }
if handle_select_client_msg(&text, &source, &chat_hub).await { continue; }
// ── /sethome ──────────────────────────────────────────────────
let client_msg: ClientMessage = match serde_json::from_str(&text) {
@@ -146,7 +166,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
let cmd = client_msg.content.trim();
if cmd == "/sethome" {
let msg = match skald.chat_hub().set_home(&source).await {
let msg = match chat_hub.set_home(&source).await {
Ok(_) => "🏠 Web impostato come **home**. Le notifiche degli agenti arriveranno qui.".to_string(),
Err(e) => format!("⚠️ Errore: {e}"),
};
@@ -172,7 +192,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
}
if cmd == "/context" {
match skald.chat_hub().context_info(&source).await {
match chat_hub.context_info(&source).await {
Ok((input, output)) => {
let input_str = input.map_or("?".to_string(), |t| t.to_string());
let output_str = output.map_or("?".to_string(), |t| t.to_string());
@@ -192,7 +212,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
}
if cmd == "/cost" {
match skald.chat_hub().cost_info(&source).await {
match chat_hub.cost_info(&source).await {
Ok(Some(c)) => {
let _ = socket.send(to_msg(&ServerEvent::Done {
message_id: 0,
@@ -219,7 +239,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
}
if cmd == "/compact" {
match skald.chat_hub().force_compact(&source).await {
match chat_hub.force_compact(&source).await {
Ok(true) => {
let _ = socket.send(to_msg(&ServerEvent::Done {
message_id: 0,
@@ -246,7 +266,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
}
if cmd == "/resettools" {
match skald.chat_hub().reset_mcp(&source).await {
match chat_hub.reset_mcp(&source).await {
Ok(()) => {
let _ = socket.send(to_msg(&ServerEvent::Done {
message_id: 0,
@@ -264,7 +284,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
}
if cmd == "/models" {
let items = skald.chat_hub().list_clients_marked(&source).await;
let items = chat_hub.list_clients_marked(&source).await;
let content = format_models_md(&items);
let _ = socket.send(to_msg(&ServerEvent::Done {
message_id: 0,
@@ -277,7 +297,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
}
if let Some(arg) = cmd.strip_prefix("/model").map(str::trim) {
let outcome = skald.chat_hub().apply_model_command(&source, arg).await;
let outcome = chat_hub.apply_model_command(&source, arg).await;
let content = match outcome {
ModelCommandOutcome::Set(name) => format!("✅ Model set: **{name}**"),
ModelCommandOutcome::Cleared => "✅ Model reset to **auto**.".to_string(),
@@ -353,14 +373,14 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
// client lives in ChatHub.selected_clients[source]. The web
// `/model` command and the dropdown both flow through
// set_selected_client, which broadcasts ClientSelected.
client_name: skald.chat_hub().get_selected_client(&source).await,
client_name: chat_hub.get_selected_client(&source).await,
extra_system_context: Some(WEB_FORMAT_CONTEXT.to_string()),
// SPA-only tool: lets the assistant open a file in the user's
// viewer. Injected here (not in the registry) so it exists only
// for ws.rs clients (web + mobile), never for the Telegram plugin.
interface_tools: vec![
skald_core::tools::show_file::make_tool(
Arc::clone(skald.chat_hub()),
Arc::clone(&chat_hub),
source.clone(),
),
],
@@ -368,7 +388,7 @@ async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, source: String)
};
// send_message only enqueues — the turn runs on ChatHub's per-source
// consumer — so awaiting inline keeps this WS read loop responsive.
if let Err(e) = skald.chat_hub().send_message(&source, &content, opts).await {
if let Err(e) = chat_hub.send_message(&source, &content, opts).await {
tracing::error!(error = %e, source = %source, "send_message enqueue failed");
}
}
src/frontend/api/ws_session.rs
+14 -4
View File
@@ -6,24 +6,34 @@ use axum::{
ws::{Message, WebSocket, WebSocketUpgrade},
},
response::IntoResponse,
Extension,
};
use tokio::sync::broadcast;
use tracing::{info, warn};
use skald_core::skald::Skald;
use super::guard::AuthUser;
pub async fn handler(
ws: WebSocketUpgrade,
Path(id): Path<i64>,
Extension(auth): Extension<AuthUser>,
State(skald): State<Arc<Skald>>,
) -> impl IntoResponse {
ws.on_upgrade(move |socket| handle_socket(socket, skald, id))
ws.on_upgrade(move |socket| handle_socket(socket, skald, id, auth.user_id))
}
async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, session_id: i64) {
info!(session_id, "session-watch WS connected");
async fn handle_socket(mut socket: WebSocket, skald: Arc<Skald>, session_id: i64, user_id: String) {
info!(session_id, user = %user_id, "session-watch WS connected");
let mut rx = skald.chat_hub().events("session-watch");
// Watch this user's own event stream, so a session-watch only ever sees events
// for sessions in the watcher's `{userid}.db`.
let ctx = match skald.user_context(&user_id).await {
Some(c) => c,
None => return,
};
let mut rx = ctx.chat_hub.events("session-watch");
loop {
tokio::select! {
src/main.rs
+4 -3
View File
@@ -164,15 +164,16 @@ pub async fn run_backend() -> Result<Backend> {
/// Build the plugin list. Extracted so both entry points share the same set.
fn build_plugins() -> Vec<Arc<dyn Plugin>> {
// NOTE (multi-user slice): telegram-bot, mobile-connector and honcho are
// single-user / global-ChatEventBus-subscriber plugins. They assume one user
// and cannot work under per-user isolation yet, so they are dropped from the
// build until they become multi-user-aware. See blueprint §17 and the plan.
let mut plugins: Vec<Arc<dyn Plugin>> = vec![
Arc::new(plugin_honcho::HonchoPlugin::new()),
Arc::new(plugin_telegram_bot::TelegramPlugin::new("secrets")),
Arc::new(plugin_tailscale_remote::RemotePlugin::new()),
Arc::new(plugin_comfyui::ComfyUIPlugin::new()),
Arc::new(plugin_tts_orpheus_3b::OrpheusTtsPlugin::new()),
Arc::new(plugin_tts_kokoro::KokoroTtsPlugin::new()),
Arc::new(plugin_elevenlabs::ElevenLabsPlugin::new()),
Arc::new(plugin_mobile_connector::MobileConnectorPlugin::new()),
];
#[cfg(feature = "whisper-local")]
plugins.push(Arc::new(plugin_transcribe_whisper_local::WhisperLocalPlugin::new()));