From d533d0a30e233666fe77b278090d16bf1391e3b0 Mon Sep 17 00:00:00 2001 From: XiuChengWu <732857315@qq.com> Date: Wed, 10 Jun 2026 01:26:46 +0800 Subject: [PATCH] Refactor syscall modules to use async/await patterns - Updated `DisplaySyscall` to use `tokio::io::AsyncWriteExt` for asynchronous text output. - Refactored `FsSyscall` to read files asynchronously with offset and length handling. - Modified `MemorySyscall` to use `tokio::task::spawn_blocking` for database operations, allowing async access to SQLite. - Enhanced `NetworkSyscall` to utilize `reqwest` for HTTP requests, replacing the previous `curl` command execution. - Improved `ProcessSyscall` to manage subprocesses with async tasks for stdin, stdout, and stderr handling. - Updated `TimerSyscall` to simplify timer management. - Adjusted plugin implementations for better async support and error handling. - Added `tokio-stream` and `tracing-subscriber` dependencies to `Cargo.toml` for enhanced async stream handling and logging. --- Cargo.lock | 726 +++++++++++++++++++++++- Cargo.toml | 2 + core/agentsd/Cargo.toml | 2 + core/agentsd/src/callback.rs | 20 +- core/agentsd/src/main.rs | 8 +- core/agentsd/src/paths.rs | 5 +- core/agentsd/src/permission.rs | 9 +- core/agentsd/src/server.rs | 821 ++++++++++++++++++++++------ core/agentsd/src/syscall/display.rs | 38 +- core/agentsd/src/syscall/fs.rs | 35 +- core/agentsd/src/syscall/memory.rs | 176 +++--- core/agentsd/src/syscall/mod.rs | 6 +- core/agentsd/src/syscall/network.rs | 104 ++-- core/agentsd/src/syscall/process.rs | 276 ++++++++-- core/agentsd/src/syscall/timer.rs | 10 +- plugins/ai_test/src/main.rs | 21 +- plugins/claudecode/Cargo.toml | 2 + plugins/claudecode/src/main.rs | 94 ++-- plugins/echo/src/main.rs | 11 +- plugins/hermes/src/main.rs | 13 +- plugins/plugin-sdk/src/lib.rs | 7 +- 21 files changed, 1950 insertions(+), 436 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4cf4e1a..31433c9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8,8 +8,10 @@ version = "0.1.0" dependencies = [ "agentsd-proto", "anyhow", + "libc", "notify", "prost", + "reqwest", "rusqlite", "serde", "serde_json", @@ -21,6 +23,76 @@ dependencies = [ "uuid", ] +[[package]] +name = "agentsd-plugin-ai-test" +version = "0.1.0" +dependencies = [ + "agentsd-plugin-sdk", + "agentsd-proto", + "anyhow", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "tonic", +] + +[[package]] +name = "agentsd-plugin-claudecode" +version = "2.1.169" +dependencies = [ + "agentsd-plugin-sdk", + "agentsd-proto", + "anyhow", + "futures-core", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "tonic", + "tracing", + "tracing-subscriber", +] + +[[package]] +name = "agentsd-plugin-echo" +version = "0.1.0" +dependencies = [ + "agentsd-plugin-sdk", + "agentsd-proto", + "anyhow", + "tokio", + "tonic", +] + +[[package]] +name = "agentsd-plugin-hermes" +version = "0.16.0" +dependencies = [ + "agentsd-plugin-sdk", + "agentsd-proto", + "anyhow", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "tonic", +] + +[[package]] +name = "agentsd-plugin-sdk" +version = "0.1.0" +dependencies = [ + "agentsd-proto", + "anyhow", + "serde", + "serde_json", + "tokio", + "tokio-stream", + "tonic", + "tracing", +] + [[package]] name = "agentsd-proto" version = "0.1.0" @@ -195,6 +267,23 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + +[[package]] +name = "displaydoc" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "either" version = "1.16.0" @@ -269,6 +358,15 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + [[package]] name = "fsevent-sys" version = "4.1.0" @@ -324,8 +422,24 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ "cfg-if", + "js-sys", "libc", "wasi", + "wasm-bindgen", +] + +[[package]] +name = "getrandom" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" +dependencies = [ + "cfg-if", + "js-sys", + "libc", + "r-efi 5.3.0", + "wasip2", + "wasm-bindgen", ] [[package]] @@ -336,7 +450,7 @@ checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" dependencies = [ "cfg-if", "libc", - "r-efi", + "r-efi 6.0.0", "wasip2", "wasip3", ] @@ -472,6 +586,22 @@ dependencies = [ "want", ] +[[package]] +name = "hyper-rustls" +version = "0.27.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33ca68d021ef39cf6463ab54c1d0f5daf03377b70561305bb89a8f83aab66e0f" +dependencies = [ + "http", + "hyper", + "hyper-util", + "rustls", + "tokio", + "tokio-rustls", + "tower-service", + "webpki-roots", +] + [[package]] name = "hyper-timeout" version = "0.5.2" @@ -491,13 +621,16 @@ version = "0.1.20" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "96547c2556ec9d12fb1578c4eaf448b04993e7fb79cbaad930a656880a6bdfa0" dependencies = [ + "base64", "bytes", "futures-channel", "futures-util", "http", "http-body", "hyper", + "ipnet", "libc", + "percent-encoding", "pin-project-lite", "socket2 0.6.4", "tokio", @@ -505,12 +638,115 @@ dependencies = [ "tracing", ] +[[package]] +name = "icu_collections" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2984d1cd16c883d7935b9e07e44071dca8d917fd52ecc02c04d5fa0b5a3f191c" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92219b62b3e2b4d88ac5119f8904c10f8f61bf7e95b640d25ba3075e6cac2c29" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c56e5ee99d6e3d33bd91c5d85458b6005a22140021cc324cea84dd0e72cff3b4" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da3be0ae77ea334f4da67c12f149704f19f81d1adf7c51cf482943e84a2bad38" + +[[package]] +name = "icu_properties" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bee3b67d0ea5c2cca5003417989af8996f8604e34fb9ddf96208a033901e70de" +dependencies = [ + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e2bbb201e0c04f7b4b3e14382af113e17ba4f63e2c9d2ee626b720cbce54a14" + +[[package]] +name = "icu_provider" +version = "2.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "139c4cf31c8b5f33d7e199446eff9c1e02decfc2f0eec2c8d71f65befa45b421" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + [[package]] name = "id-arena" version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + [[package]] name = "indexmap" version = "1.9.3" @@ -562,6 +798,12 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "ipnet" +version = "2.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" + [[package]] name = "itertools" version = "0.14.0" @@ -643,6 +885,12 @@ version = "0.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53" +[[package]] +name = "litemap" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0" + [[package]] name = "lock_api" version = "0.4.14" @@ -658,6 +906,12 @@ version = "0.4.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "953f07c43838f8e6f9758cab68bf5bed85465e7587ebe0b823f1bcd81978ad3a" +[[package]] +name = "lru-slab" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" + [[package]] name = "matchers" version = "0.2.0" @@ -817,6 +1071,15 @@ version = "0.3.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" +[[package]] +name = "potential_utf" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0103b1cef7ec0cf76490e969665504990193874ea05c85ff9bab8b911d0a0564" +dependencies = [ + "zerovec", +] + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -897,6 +1160,61 @@ dependencies = [ "prost", ] +[[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 0.6.4", + "thiserror", + "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", + "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 0.6.4", + "tracing", + "windows-sys 0.52.0", +] + [[package]] name = "quote" version = "1.0.45" @@ -906,6 +1224,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "r-efi" version = "6.0.0" @@ -919,8 +1243,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" dependencies = [ "libc", - "rand_chacha", - "rand_core", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +dependencies = [ + "rand_chacha 0.9.0", + "rand_core 0.9.5", ] [[package]] @@ -930,7 +1264,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "rand_chacha" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" +dependencies = [ + "ppv-lite86", + "rand_core 0.9.5", ] [[package]] @@ -942,6 +1286,15 @@ dependencies = [ "getrandom 0.2.17", ] +[[package]] +name = "rand_core" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" +dependencies = [ + "getrandom 0.3.4", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -980,6 +1333,58 @@ 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", + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-rustls", + "hyper-util", + "js-sys", + "log", + "percent-encoding", + "pin-project-lite", + "quinn", + "rustls", + "rustls-pki-types", + "serde", + "serde_json", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tokio-rustls", + "tower 0.5.3", + "tower-http", + "tower-service", + "url", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", + "webpki-roots", +] + +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.17", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rusqlite" version = "0.32.1" @@ -994,6 +1399,12 @@ dependencies = [ "smallvec", ] +[[package]] +name = "rustc-hash" +version = "2.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94300abf3f1ae2e2b8ffb7b58043de3d399c73fa6f4b73826402a5c457614dbe" + [[package]] name = "rustix" version = "1.1.4" @@ -1007,12 +1418,53 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "rustls" +version = "0.23.40" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef86cd5876211988985292b91c96a8f2d298df24e75989a43a3c73f2d4d8168b" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-pki-types" +version = "1.14.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9" +dependencies = [ + "web-time", + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61c429a8649f110dddef65e2a5ad240f747e85f7758a6bccc7e5777bd33f756e" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + [[package]] name = "same-file" version = "1.0.6" @@ -1077,6 +1529,18 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -1134,6 +1598,18 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + +[[package]] +name = "subtle" +version = "2.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" + [[package]] name = "syn" version = "2.0.117" @@ -1150,6 +1626,20 @@ name = "sync_wrapper" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" +dependencies = [ + "futures-core", +] + +[[package]] +name = "synstructure" +version = "0.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] [[package]] name = "tempfile" @@ -1164,6 +1654,26 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "thread_local" version = "1.1.9" @@ -1173,6 +1683,31 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "tinystr" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8323304221c2a851516f22236c5722a72eaa19749016521d6dff0824447d96d" +dependencies = [ + "displaydoc", + "zerovec", +] + +[[package]] +name = "tinyvec" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e61e67053d25a4e82c844e8424039d9745781b3fc4f32b8d55ed50f5f667ef3" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.52.3" @@ -1201,6 +1736,16 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio-rustls" +version = "0.26.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1729aa945f29d91ba541258c8df89027d5792d85a8841fb65e8bf0f4ede4ef61" +dependencies = [ + "rustls", + "tokio", +] + [[package]] name = "tokio-stream" version = "0.1.18" @@ -1280,7 +1825,7 @@ dependencies = [ "indexmap 1.9.3", "pin-project", "pin-project-lite", - "rand", + "rand 0.8.6", "slab", "tokio", "tokio-util", @@ -1299,10 +1844,29 @@ dependencies = [ "futures-util", "pin-project-lite", "sync_wrapper", + "tokio", "tower-layer", "tower-service", ] +[[package]] +name = "tower-http" +version = "0.6.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" +dependencies = [ + "bitflags 2.13.0", + "bytes", + "futures-util", + "http", + "http-body", + "pin-project-lite", + "tower 0.5.3", + "tower-layer", + "tower-service", + "url", +] + [[package]] name = "tower-layer" version = "0.3.3" @@ -1394,6 +1958,30 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" + [[package]] name = "uuid" version = "1.23.2" @@ -1479,6 +2067,16 @@ dependencies = [ "wasm-bindgen-shared", ] +[[package]] +name = "wasm-bindgen-futures" +version = "0.4.73" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54568702fabf5d4849ce2b90fadfa64168a097eaf4b351ce9df8b687a0086aaf" +dependencies = [ + "js-sys", + "wasm-bindgen", +] + [[package]] name = "wasm-bindgen-macro" version = "0.2.123" @@ -1545,6 +2143,35 @@ dependencies = [ "semver", ] +[[package]] +name = "web-sys" +version = "0.3.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e0871acf327f283dc6da28a1696cdc64fb355ba9f935d052021fa77f35cce69" +dependencies = [ + "js-sys", + "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 = "webpki-roots" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52f5ee44c96cf55f1b349600768e3ece3a8f26010c05265ab73f945bb1a2eb9d" +dependencies = [ + "rustls-pki-types", +] + [[package]] name = "winapi-util" version = "0.1.11" @@ -1736,6 +2363,35 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "writeable" +version = "0.6.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" + +[[package]] +name = "yoke" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "709fe23a0424b6a435d82152b1bd3fdfb0833487d5fa90d05d42762a9891fef5" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "synstructure", +] + [[package]] name = "zerocopy" version = "0.8.50" @@ -1756,6 +2412,66 @@ dependencies = [ "syn", ] +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" +dependencies = [ + "proc-macro2", + "quote", + "syn", + "synstructure", +] + +[[package]] +name = "zeroize" +version = "1.8.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" + +[[package]] +name = "zerotrie" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0f9152d31db0792fa83f70fb2f83148effb5c1f5b8c7686c3459e361d9bc20bf" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "90f911cbc359ab6af17377d242225f4d75119aec87ea711a880987b18cd7b239" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "zmij" version = "1.0.21" diff --git a/Cargo.toml b/Cargo.toml index 9f88819..24939b6 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -25,3 +25,5 @@ anyhow = "1" uuid = { version = "1", features = ["v4"] } notify = "7" futures-core = "0.3" +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls"] } +libc = "0.2" diff --git a/core/agentsd/Cargo.toml b/core/agentsd/Cargo.toml index 637f61c..dfbcd42 100644 --- a/core/agentsd/Cargo.toml +++ b/core/agentsd/Cargo.toml @@ -21,3 +21,5 @@ tracing-subscriber = { workspace = true } anyhow = { workspace = true } uuid = { workspace = true } notify = { workspace = true } +reqwest = { workspace = true } +libc = { workspace = true } diff --git a/core/agentsd/src/callback.rs b/core/agentsd/src/callback.rs index 7a1e5fb..b0b33be 100644 --- a/core/agentsd/src/callback.rs +++ b/core/agentsd/src/callback.rs @@ -14,6 +14,12 @@ pub struct CallbackBus { senders: Mutex>>, } +impl Default for CallbackBus { + fn default() -> Self { + Self::new() + } +} + impl CallbackBus { pub fn new() -> Self { Self { @@ -35,8 +41,18 @@ impl CallbackBus { pub fn fire(&self, plugin_id: &str, event: TimerEvent) { let senders = self.senders.lock().unwrap(); if let Some(tx) = senders.get(plugin_id) { - // Non-blocking send; drop if channel full - let _ = tx.try_send(event); + match tx.try_send(event) { + Ok(()) => {} + Err(mpsc::error::TrySendError::Full(_)) => { + tracing::warn!("timer event dropped: channel full for plugin={}", plugin_id); + } + Err(mpsc::error::TrySendError::Closed(_)) => { + tracing::debug!( + "timer event dropped: channel closed for plugin={}", + plugin_id + ); + } + } } } diff --git a/core/agentsd/src/main.rs b/core/agentsd/src/main.rs index e9fdfa6..10565a3 100644 --- a/core/agentsd/src/main.rs +++ b/core/agentsd/src/main.rs @@ -1,9 +1,9 @@ -mod syscall; +pub mod callback; +pub mod paths; +pub mod permission; mod plugin; mod server; -pub mod paths; -pub mod callback; -pub mod permission; +mod syscall; use anyhow::Result; use tracing_subscriber::EnvFilter; diff --git a/core/agentsd/src/paths.rs b/core/agentsd/src/paths.rs index c4e22d1..694b4a2 100644 --- a/core/agentsd/src/paths.rs +++ b/core/agentsd/src/paths.rs @@ -4,7 +4,10 @@ use std::path::PathBuf; /// Creates it if it does not exist. pub fn data_dir() -> PathBuf { let exe = std::env::current_exe().unwrap_or_else(|_| PathBuf::from(".")); - let dir = exe.parent().unwrap_or_else(|| std::path::Path::new(".")).join("data"); + let dir = exe + .parent() + .unwrap_or_else(|| std::path::Path::new(".")) + .join("data"); if !dir.exists() { std::fs::create_dir_all(&dir).ok(); } diff --git a/core/agentsd/src/permission.rs b/core/agentsd/src/permission.rs index b8d6915..b74b89c 100644 --- a/core/agentsd/src/permission.rs +++ b/core/agentsd/src/permission.rs @@ -12,13 +12,14 @@ pub fn plugin_id_from_metadata(req_metadata: &tonic::metadata::MetadataMap) -> O } /// Load a plugin and require it to be currently running. +#[allow(clippy::result_large_err)] pub fn require_running_plugin( plugins: &Arc, plugin_id: &str, ) -> Result { - let record = plugins.get(plugin_id).ok_or_else(|| { - Status::permission_denied(format!("unknown plugin: {}", plugin_id)) - })?; + let record = plugins + .get(plugin_id) + .ok_or_else(|| Status::permission_denied(format!("unknown plugin: {}", plugin_id)))?; if record.status != PluginStatus::Running { return Err(Status::permission_denied(format!( @@ -38,6 +39,7 @@ pub fn require_running_plugin( /// - `fs` grants all fs syscalls /// - `fs.*` grants all fs syscalls /// - `fs.list` grants only that method +#[allow(clippy::result_large_err)] pub fn check_permission( plugins: &Arc, req_metadata: &tonic::metadata::MetadataMap, @@ -73,6 +75,7 @@ pub fn check_permission( } /// Helper to extract plugin_id and check permission in one call within a service method. +#[allow(clippy::result_large_err)] pub fn guard( plugins: &Arc, req: &Request, diff --git a/core/agentsd/src/server.rs b/core/agentsd/src/server.rs index c9ca45a..9c33f7c 100644 --- a/core/agentsd/src/server.rs +++ b/core/agentsd/src/server.rs @@ -4,9 +4,8 @@ use crate::syscall::Kernel; use agentsd_proto::agentsd::*; use agentsd_proto::agentsd::{ audio_server::Audio as AudioService, display_server::Display as DisplayService, - fs_server::Fs as FsService, hid_server::Hid as HidService, - input_server::Input as InputService, memory_server::Memory as MemoryService, - network_server::Network as NetworkService, + fs_server::Fs as FsService, hid_server::Hid as HidService, input_server::Input as InputService, + memory_server::Memory as MemoryService, network_server::Network as NetworkService, plugin_bridge_server::PluginBridge as PluginBridgeService, process_server::Process as ProcessService, timer_server::Timer as TimerService, }; @@ -27,7 +26,12 @@ impl AgentsdServer { } } -fn auth(plugins: &Arc, req: &Request, syscall: &str) -> Result { +#[allow(clippy::result_large_err)] +fn auth( + plugins: &Arc, + req: &Request, + syscall: &str, +) -> Result { permission::guard(plugins, req, syscall) } @@ -39,47 +43,89 @@ pub struct DisplaySvc { #[tonic::async_trait] impl DisplayService for DisplaySvc { - async fn text(&self, req: Request) -> Result, Status> { + async fn text( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "display.text")?; let r = req.into_inner(); - match self.kernel.display.text(&r.content) { - Ok(()) => Ok(Response::new(DisplayResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(DisplayResponse { ok: false, error: e })), + match self.kernel.display.text(&r.content).await { + Ok(()) => Ok(Response::new(DisplayResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(DisplayResponse { + ok: false, + error: e, + })), } } - async fn rich(&self, req: Request) -> Result, Status> { + async fn rich( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "display.rich")?; let r = req.into_inner(); - match self.kernel.display.rich(&r.markup, &r.format) { - Ok(()) => Ok(Response::new(DisplayResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(DisplayResponse { ok: false, error: e })), + match self.kernel.display.rich(&r.markup, &r.format).await { + Ok(()) => Ok(Response::new(DisplayResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(DisplayResponse { + ok: false, + error: e, + })), } } - async fn image(&self, req: Request) -> Result, Status> { + async fn image( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "display.image")?; let r = req.into_inner(); - match self.kernel.display.image(&r.data, &r.format) { - Ok(()) => Ok(Response::new(DisplayResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(DisplayResponse { ok: false, error: e })), + match self.kernel.display.image(&r.data, &r.format).await { + Ok(()) => Ok(Response::new(DisplayResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(DisplayResponse { + ok: false, + error: e, + })), } } async fn clear(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "display.clear")?; - match self.kernel.display.clear() { - Ok(()) => Ok(Response::new(DisplayResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(DisplayResponse { ok: false, error: e })), + match self.kernel.display.clear().await { + Ok(()) => Ok(Response::new(DisplayResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(DisplayResponse { + ok: false, + error: e, + })), } } - async fn notify(&self, req: Request) -> Result, Status> { + async fn notify( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "display.notify")?; let r = req.into_inner(); - match self.kernel.display.notify(&r.message) { - Ok(()) => Ok(Response::new(DisplayResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(DisplayResponse { ok: false, error: e })), + match self.kernel.display.notify(&r.message).await { + Ok(()) => Ok(Response::new(DisplayResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(DisplayResponse { + ok: false, + error: e, + })), } } } @@ -92,7 +138,10 @@ pub struct AudioSvc { #[tonic::async_trait] impl AudioService for AudioSvc { - async fn play(&self, req: Request>) -> Result, Status> { + async fn play( + &self, + req: Request>, + ) -> Result, Status> { auth(&self.plugins, &req, "audio.play")?; use tokio_stream::StreamExt; let mut stream = req.into_inner(); @@ -106,16 +155,28 @@ impl AudioService for AudioSvc { all_data.extend(chunk.data); } match self.kernel.audio.play(&all_data, &format) { - Ok(()) => Ok(Response::new(AudioResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(AudioResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(AudioResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(AudioResponse { + ok: false, + error: e, + })), } } async fn stop(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "audio.stop")?; match self.kernel.audio.stop() { - Ok(()) => Ok(Response::new(AudioResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(AudioResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(AudioResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(AudioResponse { + ok: false, + error: e, + })), } } @@ -123,34 +184,57 @@ impl AudioService for AudioSvc { auth(&self.plugins, &req, "audio.volume")?; let r = req.into_inner(); match self.kernel.audio.volume(r.level) { - Ok(()) => Ok(Response::new(AudioResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(AudioResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(AudioResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(AudioResponse { + ok: false, + error: e, + })), } } - async fn record_start(&self, req: Request) -> Result, Status> { + async fn record_start( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "audio.record_start")?; let r = req.into_inner(); match self.kernel.audio.record_start(&r.format) { - Ok(()) => Ok(Response::new(AudioResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(AudioResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(AudioResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(AudioResponse { + ok: false, + error: e, + })), } } async fn record_stop(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "audio.record_stop")?; match self.kernel.audio.record_stop() { - Ok(data) => Ok(Response::new(AudioData { data, format: String::new() })), + Ok(data) => Ok(Response::new(AudioData { + data, + format: String::new(), + })), Err(e) => Err(Status::internal(e)), } } type RecordStreamStream = tokio_stream::wrappers::ReceiverStream>; - async fn record_stream(&self, req: Request) -> Result, Status> { + async fn record_stream( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "audio.record_stream")?; let (_tx, rx) = tokio::sync::mpsc::channel(16); - Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(rx))) + Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new( + rx, + ))) } } @@ -166,8 +250,16 @@ impl FsService for FsSvc { auth(&self.plugins, &req, "fs.read")?; let r = req.into_inner(); match self.kernel.fs.read(&r.path, r.offset, r.length).await { - Ok(data) => Ok(Response::new(FsReadResponse { data, ok: true, error: String::new() })), - Err(e) => Ok(Response::new(FsReadResponse { data: vec![], ok: false, error: e })), + Ok(data) => Ok(Response::new(FsReadResponse { + data, + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(FsReadResponse { + data: vec![], + ok: false, + error: e, + })), } } @@ -175,8 +267,14 @@ impl FsService for FsSvc { auth(&self.plugins, &req, "fs.write")?; let r = req.into_inner(); match self.kernel.fs.write(&r.path, &r.data, r.append).await { - Ok(()) => Ok(Response::new(FsResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(FsResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(FsResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(FsResponse { + ok: false, + error: e, + })), } } @@ -184,8 +282,14 @@ impl FsService for FsSvc { auth(&self.plugins, &req, "fs.delete")?; let r = req.into_inner(); match self.kernel.fs.delete(&r.path).await { - Ok(()) => Ok(Response::new(FsResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(FsResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(FsResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(FsResponse { + ok: false, + error: e, + })), } } @@ -193,8 +297,14 @@ impl FsService for FsSvc { auth(&self.plugins, &req, "fs.rename")?; let r = req.into_inner(); match self.kernel.fs.rename(&r.from, &r.to).await { - Ok(()) => Ok(Response::new(FsResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(FsResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(FsResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(FsResponse { + ok: false, + error: e, + })), } } @@ -203,13 +313,24 @@ impl FsService for FsSvc { let r = req.into_inner(); match self.kernel.fs.stat(&r.path).await { Ok(info) => Ok(Response::new(FsStatResponse { - ok: true, error: String::new(), - size: info.size, is_dir: info.is_dir, is_file: info.is_file, - modified: info.modified, created: info.created, permissions: 0, + ok: true, + error: String::new(), + size: info.size, + is_dir: info.is_dir, + is_file: info.is_file, + modified: info.modified, + created: info.created, + permissions: 0, })), Err(e) => Ok(Response::new(FsStatResponse { - ok: false, error: e, size: 0, is_dir: false, is_file: false, - modified: 0, created: 0, permissions: 0, + ok: false, + error: e, + size: 0, + is_dir: false, + is_file: false, + modified: 0, + created: 0, + permissions: 0, })), } } @@ -219,21 +340,36 @@ impl FsService for FsSvc { let r = req.into_inner(); match self.kernel.fs.list(&r.path).await { Ok(entries) => Ok(Response::new(FsListResponse { - ok: true, error: String::new(), - entries: entries.into_iter().map(|e| FsEntry { - name: e.name, is_dir: e.is_dir, size: e.size, - }).collect(), + ok: true, + error: String::new(), + entries: entries + .into_iter() + .map(|e| FsEntry { + name: e.name, + is_dir: e.is_dir, + size: e.size, + }) + .collect(), + })), + Err(e) => Ok(Response::new(FsListResponse { + ok: false, + error: e, + entries: vec![], })), - Err(e) => Ok(Response::new(FsListResponse { ok: false, error: e, entries: vec![] })), } } type WatchStream = tokio_stream::wrappers::ReceiverStream>; - async fn watch(&self, req: Request) -> Result, Status> { + async fn watch( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "fs.watch")?; let (_tx, rx) = tokio::sync::mpsc::channel(16); - Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(rx))) + Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new( + rx, + ))) } } @@ -245,52 +381,107 @@ pub struct MemorySvc { #[tonic::async_trait] impl MemoryService for MemorySvc { - async fn read(&self, req: Request) -> Result, Status> { + async fn read( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "memory.read")?; let r = req.into_inner(); - match self.kernel.memory.read(&r.key) { - Ok(Some(v)) => Ok(Response::new(MemoryReadResponse { value: v, found: true, error: String::new() })), - Ok(None) => Ok(Response::new(MemoryReadResponse { value: vec![], found: false, error: String::new() })), - Err(e) => Ok(Response::new(MemoryReadResponse { value: vec![], found: false, error: e })), - } - } - - async fn write(&self, req: Request) -> Result, Status> { - auth(&self.plugins, &req, "memory.write")?; - let r = req.into_inner(); - match self.kernel.memory.write(&r.key, &r.value) { - Ok(()) => Ok(Response::new(MemoryResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(MemoryResponse { ok: false, error: e })), - } - } - - async fn delete(&self, req: Request) -> Result, Status> { - auth(&self.plugins, &req, "memory.delete")?; - let r = req.into_inner(); - match self.kernel.memory.delete(&r.key) { - Ok(()) => Ok(Response::new(MemoryResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(MemoryResponse { ok: false, error: e })), - } - } - - async fn list(&self, req: Request) -> Result, Status> { - auth(&self.plugins, &req, "memory.list")?; - let r = req.into_inner(); - match self.kernel.memory.list(&r.prefix, r.limit) { - Ok(keys) => Ok(Response::new(MemoryListResponse { keys, error: String::new() })), - Err(e) => Ok(Response::new(MemoryListResponse { keys: vec![], error: e })), - } - } - - async fn search(&self, req: Request) -> Result, Status> { - auth(&self.plugins, &req, "memory.search")?; - let r = req.into_inner(); - match self.kernel.memory.search(&r.query, r.limit) { - Ok(hits) => Ok(Response::new(MemorySearchResponse { - hits: hits.into_iter().map(|(k, s, sc)| MemorySearchHit { key: k, snippet: s, score: sc }).collect(), + match self.kernel.memory.read(&r.key).await { + Ok(Some(v)) => Ok(Response::new(MemoryReadResponse { + value: v, + found: true, error: String::new(), })), - Err(e) => Ok(Response::new(MemorySearchResponse { hits: vec![], error: e })), + Ok(None) => Ok(Response::new(MemoryReadResponse { + value: vec![], + found: false, + error: String::new(), + })), + Err(e) => Ok(Response::new(MemoryReadResponse { + value: vec![], + found: false, + error: e, + })), + } + } + + async fn write( + &self, + req: Request, + ) -> Result, Status> { + auth(&self.plugins, &req, "memory.write")?; + let r = req.into_inner(); + match self.kernel.memory.write(&r.key, &r.value).await { + Ok(()) => Ok(Response::new(MemoryResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(MemoryResponse { + ok: false, + error: e, + })), + } + } + + async fn delete( + &self, + req: Request, + ) -> Result, Status> { + auth(&self.plugins, &req, "memory.delete")?; + let r = req.into_inner(); + match self.kernel.memory.delete(&r.key).await { + Ok(()) => Ok(Response::new(MemoryResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(MemoryResponse { + ok: false, + error: e, + })), + } + } + + async fn list( + &self, + req: Request, + ) -> Result, Status> { + auth(&self.plugins, &req, "memory.list")?; + let r = req.into_inner(); + match self.kernel.memory.list(&r.prefix, r.limit).await { + Ok(keys) => Ok(Response::new(MemoryListResponse { + keys, + error: String::new(), + })), + Err(e) => Ok(Response::new(MemoryListResponse { + keys: vec![], + error: e, + })), + } + } + + async fn search( + &self, + req: Request, + ) -> Result, Status> { + auth(&self.plugins, &req, "memory.search")?; + let r = req.into_inner(); + match self.kernel.memory.search(&r.query, r.limit).await { + Ok(hits) => Ok(Response::new(MemorySearchResponse { + hits: hits + .into_iter() + .map(|(k, s, sc)| MemorySearchHit { + key: k, + snippet: s, + score: sc, + }) + .collect(), + error: String::new(), + })), + Err(e) => Ok(Response::new(MemorySearchResponse { + hits: vec![], + error: e, + })), } } } @@ -306,41 +497,77 @@ impl NetworkService for NetworkSvc { async fn request(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "network.request")?; let r = req.into_inner(); - match self.kernel.network.request(&r.url, &r.method, &r.headers, &r.body).await { - Ok((status, headers, body)) => Ok(Response::new(HttpResponse { status, headers, body, error: String::new() })), - Err(e) => Ok(Response::new(HttpResponse { status: 0, headers: Default::default(), body: vec![], error: e })), + match self + .kernel + .network + .request(&r.url, &r.method, &r.headers, &r.body) + .await + { + Ok((status, headers, body)) => Ok(Response::new(HttpResponse { + status, + headers, + body, + error: String::new(), + })), + Err(e) => Ok(Response::new(HttpResponse { + status: 0, + headers: Default::default(), + body: vec![], + error: e, + })), } } type OpenConnStream = tokio_stream::wrappers::ReceiverStream>; - async fn open_conn(&self, req: Request) -> Result, Status> { + async fn open_conn( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "network.open_conn")?; let (_tx, rx) = tokio::sync::mpsc::channel(16); - Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(rx))) + Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new( + rx, + ))) } type ListenStream = tokio_stream::wrappers::ReceiverStream>; - async fn listen(&self, req: Request) -> Result, Status> { + async fn listen( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "network.listen")?; let (_tx, rx) = tokio::sync::mpsc::channel(16); - Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(rx))) + Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new( + rx, + ))) } async fn send(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "network.send")?; - Ok(Response::new(NetResponse { ok: false, error: "send: not yet implemented".to_string() })) + Ok(Response::new(NetResponse { + ok: false, + error: "send: not yet implemented".to_string(), + })) } async fn close(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "network.close")?; - Ok(Response::new(NetResponse { ok: false, error: "close: not yet implemented".to_string() })) + Ok(Response::new(NetResponse { + ok: false, + error: "close: not yet implemented".to_string(), + })) } - async fn available(&self, req: Request) -> Result, Status> { + async fn available( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "network.available")?; - Ok(Response::new(NetAvailableResponse { available: self.kernel.network.available() })) + Ok(Response::new(NetAvailableResponse { + available: self.kernel.network.available().await, + })) } } @@ -355,20 +582,34 @@ impl InputService for InputSvc { async fn text(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "input.text")?; match self.kernel.input.text().await { - Ok(t) => Ok(Response::new(InputTextResponse { text: t, ok: true, error: String::new() })), - Err(e) => Ok(Response::new(InputTextResponse { text: String::new(), ok: false, error: e })), + Ok(t) => Ok(Response::new(InputTextResponse { + text: t, + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(InputTextResponse { + text: String::new(), + ok: false, + error: e, + })), } } async fn key(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "input.key")?; - Ok(Response::new(InputKeyResponse { key: String::new(), action: String::new(), modifiers: vec![] })) + Err(Status::unimplemented("input.key not yet implemented")) } - async fn clipboard(&self, req: Request) -> Result, Status> { + async fn clipboard( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "input.clipboard")?; Ok(Response::new(InputClipboardResponse { - data: vec![], mime: String::new(), ok: false, error: "not implemented".to_string(), + data: vec![], + mime: String::new(), + ok: false, + error: "not implemented".to_string(), })) } @@ -377,7 +618,9 @@ impl InputService for InputSvc { async fn events(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "input.events")?; let (_tx, rx) = tokio::sync::mpsc::channel(16); - Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(rx))) + Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new( + rx, + ))) } } @@ -393,9 +636,22 @@ impl ProcessService for ProcessSvc { auth(&self.plugins, &req, "process.spawn")?; let r = req.into_inner(); let args: Vec = r.args; - match self.kernel.process.spawn(&r.cmd, &args, &r.env, &r.cwd).await { - Ok(pid) => Ok(Response::new(SpawnResponse { pid, ok: true, error: String::new() })), - Err(e) => Ok(Response::new(SpawnResponse { pid: 0, ok: false, error: e })), + match self + .kernel + .process + .spawn(&r.cmd, &args, &r.env, &r.cwd) + .await + { + Ok(pid) => Ok(Response::new(SpawnResponse { + pid, + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(SpawnResponse { + pid: 0, + ok: false, + error: e, + })), } } @@ -403,8 +659,14 @@ impl ProcessService for ProcessSvc { auth(&self.plugins, &req, "process.kill")?; let r = req.into_inner(); match self.kernel.process.kill(r.pid).await { - Ok(()) => Ok(Response::new(ProcResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(ProcResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(ProcResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(ProcResponse { + ok: false, + error: e, + })), } } @@ -412,8 +674,16 @@ impl ProcessService for ProcessSvc { auth(&self.plugins, &req, "process.wait")?; let r = req.into_inner(); match self.kernel.process.wait(r.pid).await { - Ok(code) => Ok(Response::new(WaitResponse { exit_code: code, ok: true, error: String::new() })), - Err(e) => Ok(Response::new(WaitResponse { exit_code: -1, ok: false, error: e })), + Ok(code) => Ok(Response::new(WaitResponse { + exit_code: code, + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(WaitResponse { + exit_code: -1, + ok: false, + error: e, + })), } } @@ -421,14 +691,23 @@ impl ProcessService for ProcessSvc { auth(&self.plugins, &req, "process.stdin")?; let r = req.into_inner(); match self.kernel.process.write_stdin(r.pid, &r.data).await { - Ok(()) => Ok(Response::new(ProcResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(ProcResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(ProcResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(ProcResponse { + ok: false, + error: e, + })), } } type StdoutStream = tokio_stream::wrappers::ReceiverStream>; - async fn stdout(&self, req: Request) -> Result, Status> { + async fn stdout( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "process.stdout")?; let r = req.into_inner(); let (tx, rx) = tokio::sync::mpsc::channel(64); @@ -441,7 +720,13 @@ impl ProcessService for ProcessSvc { match result { Ok(0) => break, Ok(n) => { - if tx.send(Ok(DataChunk { data: buf[..n].to_vec() })).await.is_err() { + if tx + .send(Ok(DataChunk { + data: buf[..n].to_vec(), + })) + .await + .is_err() + { break; } } @@ -449,23 +734,34 @@ impl ProcessService for ProcessSvc { } } }); - Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(rx))) + Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new( + rx, + ))) } async fn signal(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "process.signal")?; let r = req.into_inner(); match self.kernel.process.signal(r.pid, r.signal).await { - Ok(()) => Ok(Response::new(ProcResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(ProcResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(ProcResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(ProcResponse { + ok: false, + error: e, + })), } } async fn list_proc(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "process.list_proc")?; - let procs = self.kernel.process.list(); + let procs = self.kernel.process.list().await; Ok(Response::new(ProcListResponse { - procs: procs.into_iter().map(|(pid, cmd, status)| ProcInfo { pid, cmd, status }).collect(), + procs: procs + .into_iter() + .map(|(pid, cmd, status)| ProcInfo { pid, cmd, status }) + .collect(), })) } } @@ -478,32 +774,57 @@ pub struct TimerSvc { #[tonic::async_trait] impl TimerService for TimerSvc { - async fn once(&self, req: Request) -> Result, Status> { + async fn once( + &self, + req: Request, + ) -> Result, Status> { let plugin_id = auth(&self.plugins, &req, "timer.once")?; let r = req.into_inner(); let id = self.kernel.timer.once(r.delay_ms, r.callback_id, plugin_id); - Ok(Response::new(TimerResponse { timer_id: id, ok: true, error: String::new() })) - } - - async fn cron(&self, req: Request) -> Result, Status> { - auth(&self.plugins, &req, "timer.cron")?; Ok(Response::new(TimerResponse { - timer_id: String::new(), ok: false, error: "cron: not yet implemented".to_string(), + timer_id: id, + ok: true, + error: String::new(), })) } - async fn cancel(&self, req: Request) -> Result, Status> { + async fn cron( + &self, + req: Request, + ) -> Result, Status> { + auth(&self.plugins, &req, "timer.cron")?; + Ok(Response::new(TimerResponse { + timer_id: String::new(), + ok: false, + error: "cron: not yet implemented".to_string(), + })) + } + + async fn cancel( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "timer.cancel")?; let r = req.into_inner(); match self.kernel.timer.cancel(&r.timer_id) { - Ok(()) => Ok(Response::new(TimerResponse { timer_id: r.timer_id, ok: true, error: String::new() })), - Err(e) => Ok(Response::new(TimerResponse { timer_id: r.timer_id, ok: false, error: e })), + Ok(()) => Ok(Response::new(TimerResponse { + timer_id: r.timer_id, + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(TimerResponse { + timer_id: r.timer_id, + ok: false, + error: e, + })), } } async fn now(&self, req: Request) -> Result, Status> { auth(&self.plugins, &req, "timer.now")?; - Ok(Response::new(TimerNowResponse { timestamp_ms: self.kernel.timer.now() })) + Ok(Response::new(TimerNowResponse { + timestamp_ms: self.kernel.timer.now(), + })) } } @@ -515,12 +836,25 @@ pub struct HidSvc { #[tonic::async_trait] impl HidService for HidSvc { - async fn capture(&self, req: Request) -> Result, Status> { + async fn capture( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "hid.capture")?; let r = req.into_inner(); match self.kernel.hid.capture(r.x, r.y, r.width, r.height) { - Ok(image) => Ok(Response::new(CaptureResponse { image, format: "png".to_string(), ok: true, error: String::new() })), - Err(e) => Ok(Response::new(CaptureResponse { image: vec![], format: String::new(), ok: false, error: e })), + Ok(image) => Ok(Response::new(CaptureResponse { + image, + format: "png".to_string(), + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(CaptureResponse { + image: vec![], + format: String::new(), + ok: false, + error: e, + })), } } @@ -528,17 +862,32 @@ impl HidService for HidSvc { auth(&self.plugins, &req, "hid.mouse")?; let r = req.into_inner(); match self.kernel.hid.mouse(r.x, r.y, &r.action, &r.button) { - Ok(()) => Ok(Response::new(HidResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(HidResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(HidResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(HidResponse { + ok: false, + error: e, + })), } } - async fn keyboard(&self, req: Request) -> Result, Status> { + async fn keyboard( + &self, + req: Request, + ) -> Result, Status> { auth(&self.plugins, &req, "hid.keyboard")?; let r = req.into_inner(); match self.kernel.hid.keyboard(&r.key, &r.action) { - Ok(()) => Ok(Response::new(HidResponse { ok: true, error: String::new() })), - Err(e) => Ok(Response::new(HidResponse { ok: false, error: e })), + Ok(()) => Ok(Response::new(HidResponse { + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(HidResponse { + ok: false, + error: e, + })), } } @@ -546,8 +895,16 @@ impl HidService for HidSvc { auth(&self.plugins, &req, "hid.ocr")?; let r = req.into_inner(); match self.kernel.hid.ocr(r.x, r.y, r.width, r.height) { - Ok(text) => Ok(Response::new(OcrResponse { text, ok: true, error: String::new() })), - Err(e) => Ok(Response::new(OcrResponse { text: String::new(), ok: false, error: e })), + Ok(text) => Ok(Response::new(OcrResponse { + text, + ok: true, + error: String::new(), + })), + Err(e) => Ok(Response::new(OcrResponse { + text: String::new(), + ok: false, + error: e, + })), } } } @@ -560,7 +917,10 @@ pub struct BridgeSvc { #[tonic::async_trait] impl PluginBridgeService for BridgeSvc { - async fn register(&self, req: Request) -> Result, Status> { + async fn register( + &self, + req: Request, + ) -> Result, Status> { let r = req.into_inner(); let record = PluginRecord { id: r.id, @@ -569,25 +929,68 @@ impl PluginBridgeService for BridgeSvc { plugin_type: r.plugin_type, syscalls: r.syscalls, depends: r.depends, - exports: r.exports.into_iter().map(|e| (e.name, e.description)).collect(), + exports: r + .exports + .into_iter() + .map(|e| (e.name, e.description)) + .collect(), + endpoint: if r.endpoint.is_empty() { + None + } else { + Some(r.endpoint) + }, status: PluginStatus::Running, }; match self.plugins.register(record) { - Ok(()) => Ok(Response::new(RegisterResponse { ok: true, error: String::new(), session_id: uuid::Uuid::new_v4().to_string() })), - Err(e) => Ok(Response::new(RegisterResponse { ok: false, error: e, session_id: String::new() })), + Ok(()) => Ok(Response::new(RegisterResponse { + ok: true, + error: String::new(), + session_id: uuid::Uuid::new_v4().to_string(), + })), + Err(e) => Ok(Response::new(RegisterResponse { + ok: false, + error: e, + session_id: String::new(), + })), } } async fn call(&self, req: Request) -> Result, Status> { let r = req.into_inner(); let fn_name = &r.r#fn; - match self.plugins.find_export(&r.target, fn_name) { - Some(_plugin_id) => Ok(Response::new(CallResponse { - result: vec![], ok: false, - error: format!("call routing not yet implemented: {}.{}", r.target, fn_name), - })), + match self.plugins.find_export_record(&r.target, fn_name) { + Some(record) => { + let endpoint = record.endpoint.as_deref().unwrap_or_default(); + if endpoint.is_empty() { + return Ok(Response::new(CallResponse { + result: vec![], + ok: false, + error: format!("plugin '{}' has no endpoint for routing", record.id), + })); + } + let channel = tonic::transport::Channel::from_shared(endpoint.to_string()) + .map_err(|e| Status::internal(e.to_string()))? + .connect_timeout(std::time::Duration::from_secs(5)) + .timeout(std::time::Duration::from_secs(30)) + .connect() + .await + .map_err(|e| Status::unavailable(format!("connect {}: {}", endpoint, e)))?; + let mut client = + agentsd_proto::agentsd::plugin_bridge_client::PluginBridgeClient::new(channel); + let resp = client + .call(CallRequest { + target: record.id, + r#fn: fn_name.to_string(), + args: r.args, + }) + .await + .map_err(|e| Status::internal(e.to_string()))? + .into_inner(); + Ok(Response::new(resp)) + } None => Ok(Response::new(CallResponse { - result: vec![], ok: false, + result: vec![], + ok: false, error: format!("export not found: {}.{}", r.target, fn_name), })), } @@ -595,7 +998,10 @@ impl PluginBridgeService for BridgeSvc { type StreamCallStream = tokio_stream::wrappers::ReceiverStream>; - async fn stream_call(&self, req: Request) -> Result, Status> { + async fn stream_call( + &self, + req: Request, + ) -> Result, Status> { let metadata_plugin_id = permission::plugin_id_from_metadata(req.metadata()); let r = req.into_inner(); let target = r.target; @@ -611,7 +1017,9 @@ impl PluginBridgeService for BridgeSvc { let plugin_id = metadata_plugin_id.unwrap_or(target); if plugin_id.is_empty() { - return Err(Status::invalid_argument("stream_call requires target or x-plugin-id")); + return Err(Status::invalid_argument( + "stream_call requires target or x-plugin-id", + )); } permission::require_running_plugin(&self.plugins, &plugin_id)?; @@ -626,47 +1034,83 @@ impl PluginBridgeService for BridgeSvc { "timer_id": event.timer_id, "callback_id": event.callback_id, }); - let response = CallResponse { result: payload.to_string().into_bytes(), ok: true, error: String::new() }; + let response = CallResponse { + result: payload.to_string().into_bytes(), + ok: true, + error: String::new(), + }; if tx.send(Ok(response)).await.is_err() { break; } } bus.remove(&subscribed_plugin_id); }); - Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new(rx))) + Ok(Response::new(tokio_stream::wrappers::ReceiverStream::new( + rx, + ))) } - async fn heartbeat(&self, _req: Request) -> Result, Status> { + async fn heartbeat( + &self, + _req: Request, + ) -> Result, Status> { Ok(Response::new(HeartbeatResponse { ok: true })) } } fn display_svc(kernel: &Arc, plugins: &Arc) -> DisplaySvc { - DisplaySvc { kernel: kernel.clone(), plugins: plugins.clone() } + DisplaySvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } fn audio_svc(kernel: &Arc, plugins: &Arc) -> AudioSvc { - AudioSvc { kernel: kernel.clone(), plugins: plugins.clone() } + AudioSvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } fn fs_svc(kernel: &Arc, plugins: &Arc) -> FsSvc { - FsSvc { kernel: kernel.clone(), plugins: plugins.clone() } + FsSvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } fn memory_svc(kernel: &Arc, plugins: &Arc) -> MemorySvc { - MemorySvc { kernel: kernel.clone(), plugins: plugins.clone() } + MemorySvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } fn network_svc(kernel: &Arc, plugins: &Arc) -> NetworkSvc { - NetworkSvc { kernel: kernel.clone(), plugins: plugins.clone() } + NetworkSvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } fn input_svc(kernel: &Arc, plugins: &Arc) -> InputSvc { - InputSvc { kernel: kernel.clone(), plugins: plugins.clone() } + InputSvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } fn process_svc(kernel: &Arc, plugins: &Arc) -> ProcessSvc { - ProcessSvc { kernel: kernel.clone(), plugins: plugins.clone() } + ProcessSvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } fn timer_svc(kernel: &Arc, plugins: &Arc) -> TimerSvc { - TimerSvc { kernel: kernel.clone(), plugins: plugins.clone() } + TimerSvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } fn hid_svc(kernel: &Arc, plugins: &Arc) -> HidSvc { - HidSvc { kernel: kernel.clone(), plugins: plugins.clone() } + HidSvc { + kernel: kernel.clone(), + plugins: plugins.clone(), + } } // ===== Server startup ===== @@ -685,16 +1129,31 @@ pub async fn run(kernel: Kernel) -> anyhow::Result<()> { tracing::info!("agentsd listening on {}", addr); Server::builder() - .add_service(DisplayServer::new(display_svc(&server.kernel, &server.plugins))) + .add_service(DisplayServer::new(display_svc( + &server.kernel, + &server.plugins, + ))) .add_service(AudioServer::new(audio_svc(&server.kernel, &server.plugins))) .add_service(FsServer::new(fs_svc(&server.kernel, &server.plugins))) - .add_service(MemoryServer::new(memory_svc(&server.kernel, &server.plugins))) - .add_service(NetworkServer::new(network_svc(&server.kernel, &server.plugins))) + .add_service(MemoryServer::new(memory_svc( + &server.kernel, + &server.plugins, + ))) + .add_service(NetworkServer::new(network_svc( + &server.kernel, + &server.plugins, + ))) .add_service(InputServer::new(input_svc(&server.kernel, &server.plugins))) - .add_service(ProcessServer::new(process_svc(&server.kernel, &server.plugins))) + .add_service(ProcessServer::new(process_svc( + &server.kernel, + &server.plugins, + ))) .add_service(TimerServer::new(timer_svc(&server.kernel, &server.plugins))) .add_service(HidServer::new(hid_svc(&server.kernel, &server.plugins))) - .add_service(PluginBridgeServer::new(BridgeSvc { kernel: server.kernel.clone(), plugins: server.plugins.clone() })) + .add_service(PluginBridgeServer::new(BridgeSvc { + kernel: server.kernel.clone(), + plugins: server.plugins.clone(), + })) .serve_with_shutdown(addr, async { tokio::signal::ctrl_c().await.ok(); tracing::info!("shutdown signal received, stopping..."); diff --git a/core/agentsd/src/syscall/display.rs b/core/agentsd/src/syscall/display.rs index 00dcaf4..f1a4e07 100644 --- a/core/agentsd/src/syscall/display.rs +++ b/core/agentsd/src/syscall/display.rs @@ -1,4 +1,4 @@ -use std::io::Write as IoWrite; +use tokio::io::AsyncWriteExt; /// Display syscall: output to user. pub struct DisplaySyscall; @@ -8,32 +8,34 @@ impl DisplaySyscall { Self } - pub fn text(&self, content: &str) -> Result<(), String> { - let stdout = std::io::stdout(); - let mut handle = stdout.lock(); - handle + pub async fn text(&self, content: &str) -> Result<(), String> { + let mut stdout = tokio::io::stdout(); + stdout .write_all(content.as_bytes()) + .await .map_err(|e| e.to_string())?; - handle.write_all(b"\n").map_err(|e| e.to_string())?; - handle.flush().map_err(|e| e.to_string()) + stdout.write_all(b"\n").await.map_err(|e| e.to_string())?; + stdout.flush().await.map_err(|e| e.to_string()) } - pub fn rich(&self, markup: &str, _format: &str) -> Result<(), String> { - // For now, render as plain text. Rich rendering is a system plugin concern. - self.text(markup) + pub async fn rich(&self, markup: &str, _format: &str) -> Result<(), String> { + self.text(markup).await } - pub fn image(&self, _data: &[u8], _format: &str) -> Result<(), String> { - self.text("[image]") + pub async fn image(&self, _data: &[u8], _format: &str) -> Result<(), String> { + self.text("[image]").await } - pub fn clear(&self) -> Result<(), String> { - // ANSI clear - print!("\x1b[2J\x1b[H"); - std::io::stdout().flush().map_err(|e| e.to_string()) + pub async fn clear(&self) -> Result<(), String> { + let mut stdout = tokio::io::stdout(); + stdout + .write_all(b"\x1b[2J\x1b[H") + .await + .map_err(|e| e.to_string())?; + stdout.flush().await.map_err(|e| e.to_string()) } - pub fn notify(&self, message: &str) -> Result<(), String> { - self.text(&format!("[notify] {}", message)) + pub async fn notify(&self, message: &str) -> Result<(), String> { + self.text(&format!("[notify] {}", message)).await } } diff --git a/core/agentsd/src/syscall/fs.rs b/core/agentsd/src/syscall/fs.rs index eeaed3e..8916164 100644 --- a/core/agentsd/src/syscall/fs.rs +++ b/core/agentsd/src/syscall/fs.rs @@ -1,5 +1,8 @@ use std::path::Path; use tokio::fs as async_fs; +use tokio::io::{AsyncReadExt, AsyncSeekExt}; + +const MAX_READ_SIZE: usize = 64 * 1024 * 1024; // 64 MB /// FS syscall: file system operations. pub struct FsSyscall; @@ -10,17 +13,25 @@ impl FsSyscall { } pub async fn read(&self, path: &str, offset: i64, length: i64) -> Result, String> { - let data = async_fs::read(path).await.map_err(|e| e.to_string())?; - let start = if offset > 0 { offset as usize } else { 0 }; - let end = if length > 0 { - std::cmp::min(start + length as usize, data.len()) - } else { - data.len() - }; - if start >= data.len() { - return Ok(Vec::new()); + let mut file = tokio::fs::File::open(path) + .await + .map_err(|e| e.to_string())?; + if offset > 0 { + file.seek(std::io::SeekFrom::Start(offset as u64)) + .await + .map_err(|e| e.to_string())?; } - Ok(data[start..end].to_vec()) + let len = if length > 0 { + std::cmp::min(length as usize, MAX_READ_SIZE) + } else { + let meta = file.metadata().await.map_err(|e| e.to_string())?; + let remaining = (meta.len() as usize).saturating_sub(offset.max(0) as usize); + std::cmp::min(remaining, MAX_READ_SIZE) + }; + let mut buf = vec![0u8; len]; + let n = file.read(&mut buf).await.map_err(|e| e.to_string())?; + buf.truncate(n); + Ok(buf) } pub async fn write(&self, path: &str, data: &[u8], append: bool) -> Result<(), String> { @@ -50,9 +61,7 @@ impl FsSyscall { .await .map_err(|e| e.to_string()) } else { - async_fs::remove_file(path) - .await - .map_err(|e| e.to_string()) + async_fs::remove_file(path).await.map_err(|e| e.to_string()) } } diff --git a/core/agentsd/src/syscall/memory.rs b/core/agentsd/src/syscall/memory.rs index e2cf667..c8bf586 100644 --- a/core/agentsd/src/syscall/memory.rs +++ b/core/agentsd/src/syscall/memory.rs @@ -1,10 +1,10 @@ use rusqlite::{params, Connection}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; /// Memory syscall: structured KV storage with FTS5 full-text search. /// Data is persisted to `/data/memory.db`. pub struct MemorySyscall { - conn: Mutex, + conn: Arc>, } impl MemorySyscall { @@ -12,7 +12,9 @@ impl MemorySyscall { let db_path = crate::paths::data_dir().join("memory.db"); let conn = Connection::open(&db_path)?; conn.execute_batch( - "CREATE TABLE IF NOT EXISTS kv ( + "PRAGMA journal_mode=WAL; + PRAGMA busy_timeout=5000; + CREATE TABLE IF NOT EXISTS kv ( key TEXT PRIMARY KEY, value BLOB NOT NULL, text_value TEXT, @@ -34,78 +36,114 @@ impl MemorySyscall { )?; tracing::info!("memory: opened {}", db_path.display()); Ok(Self { - conn: Mutex::new(conn), + conn: Arc::new(Mutex::new(conn)), }) } - pub fn read(&self, key: &str) -> Result>, String> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn - .prepare("SELECT value FROM kv WHERE key = ?1") - .map_err(|e| e.to_string())?; - let result = stmt - .query_row(params![key], |row| row.get::<_, Vec>(0)) - .ok(); - Ok(result) + pub async fn read(&self, key: &str) -> Result>, String> { + let conn = self.conn.clone(); + let key = key.to_string(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().unwrap(); + let mut stmt = conn + .prepare("SELECT value FROM kv WHERE key = ?1") + .map_err(|e| e.to_string())?; + let result = stmt + .query_row(params![key], |row| row.get::<_, Vec>(0)) + .ok(); + Ok(result) + }) + .await + .map_err(|e| e.to_string())? } - pub fn write(&self, key: &str, value: &[u8]) -> Result<(), String> { - let conn = self.conn.lock().unwrap(); - let text_value = String::from_utf8(value.to_vec()).ok(); - conn.execute( - "INSERT OR REPLACE INTO kv (key, value, text_value, updated_at) VALUES (?1, ?2, ?3, strftime('%s','now'))", - params![key, value, text_value], - ) - .map_err(|e| e.to_string())?; - Ok(()) - } - - pub fn delete(&self, key: &str) -> Result<(), String> { - let conn = self.conn.lock().unwrap(); - conn.execute("DELETE FROM kv WHERE key = ?1", params![key]) - .map_err(|e| e.to_string())?; - Ok(()) - } - - pub fn list(&self, prefix: &str, limit: i32) -> Result, String> { - let conn = self.conn.lock().unwrap(); - let lim = if limit > 0 { limit } else { 1000 }; - let mut stmt = conn - .prepare("SELECT key FROM kv WHERE key LIKE ?1 ORDER BY key LIMIT ?2") - .map_err(|e| e.to_string())?; - let pattern = format!("{}%", prefix); - let rows = stmt - .query_map(params![pattern, lim], |row| row.get::<_, String>(0)) - .map_err(|e| e.to_string())?; - let mut keys = Vec::new(); - for row in rows { - keys.push(row.map_err(|e| e.to_string())?); - } - Ok(keys) - } - - pub fn search(&self, query: &str, limit: i32) -> Result, String> { - let conn = self.conn.lock().unwrap(); - let lim = if limit > 0 { limit } else { 20 }; - let mut stmt = conn - .prepare( - "SELECT key, snippet(kv_fts, 1, '', '', '...', 32), rank - FROM kv_fts WHERE kv_fts MATCH ?1 ORDER BY rank LIMIT ?2", + pub async fn write(&self, key: &str, value: &[u8]) -> Result<(), String> { + let conn = self.conn.clone(); + let key = key.to_string(); + let value = value.to_vec(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().unwrap(); + let text_value: Option = + std::str::from_utf8(&value).ok().map(str::to_owned); + conn.execute( + "INSERT OR REPLACE INTO kv (key, value, text_value, updated_at) VALUES (?1, ?2, ?3, strftime('%s','now'))", + params![key, value, text_value], ) .map_err(|e| e.to_string())?; - let rows = stmt - .query_map(params![query, lim], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, String>(1)?, - row.get::<_, f64>(2)? as f32, - )) - }) - .map_err(|e| e.to_string())?; - let mut hits = Vec::new(); - for row in rows { - hits.push(row.map_err(|e| e.to_string())?); - } - Ok(hits) + Ok(()) + }) + .await + .map_err(|e| e.to_string())? + } + + pub async fn delete(&self, key: &str) -> Result<(), String> { + let conn = self.conn.clone(); + let key = key.to_string(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().unwrap(); + conn.execute("DELETE FROM kv WHERE key = ?1", params![key]) + .map_err(|e| e.to_string())?; + Ok(()) + }) + .await + .map_err(|e| e.to_string())? + } + + pub async fn list(&self, prefix: &str, limit: i32) -> Result, String> { + let conn = self.conn.clone(); + let prefix = prefix.to_string(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().unwrap(); + let lim = if limit > 0 { limit } else { 1000 }; + let mut stmt = conn + .prepare("SELECT key FROM kv WHERE key LIKE ?1 ORDER BY key LIMIT ?2") + .map_err(|e| e.to_string())?; + let pattern = format!("{}%", prefix); + let rows = stmt + .query_map(params![pattern, lim], |row| row.get::<_, String>(0)) + .map_err(|e| e.to_string())?; + let mut keys = Vec::new(); + for row in rows { + keys.push(row.map_err(|e| e.to_string())?); + } + Ok(keys) + }) + .await + .map_err(|e| e.to_string())? + } + + pub async fn search( + &self, + query: &str, + limit: i32, + ) -> Result, String> { + let conn = self.conn.clone(); + let query = query.to_string(); + tokio::task::spawn_blocking(move || { + let conn = conn.lock().unwrap(); + let lim = if limit > 0 { limit } else { 20 }; + let mut stmt = conn + .prepare( + "SELECT key, snippet(kv_fts, 1, '', '', '...', 32), rank + FROM kv_fts WHERE kv_fts MATCH ?1 ORDER BY rank LIMIT ?2", + ) + .map_err(|e| e.to_string())?; + let rows = stmt + .query_map(params![query, lim], |row| { + Ok(( + row.get::<_, String>(0)?, + row.get::<_, String>(1)?, + row.get::<_, f64>(2)? as f32, + )) + }) + .map_err(|e| e.to_string())?; + let mut hits = Vec::new(); + for row in rows { + hits.push(row.map_err(|e| e.to_string())?); + } + Ok(hits) + }) + .await + .map_err(|e| e.to_string())? } } diff --git a/core/agentsd/src/syscall/mod.rs b/core/agentsd/src/syscall/mod.rs index 1b65918..801f9a2 100644 --- a/core/agentsd/src/syscall/mod.rs +++ b/core/agentsd/src/syscall/mod.rs @@ -1,12 +1,12 @@ -pub mod display; pub mod audio; +pub mod display; pub mod fs; +pub mod hid; +pub mod input; pub mod memory; pub mod network; pub mod process; pub mod timer; -pub mod hid; -pub mod input; use anyhow::Result; use std::sync::Arc; diff --git a/core/agentsd/src/syscall/network.rs b/core/agentsd/src/syscall/network.rs index 8cf70ed..85c5e42 100644 --- a/core/agentsd/src/syscall/network.rs +++ b/core/agentsd/src/syscall/network.rs @@ -1,60 +1,94 @@ +use std::collections::HashMap; +use std::net::SocketAddr; +use std::time::Duration; + +const MAX_RESPONSE_BODY: u64 = 16 * 1024 * 1024; // 16 MB + /// Network syscall: remote communication. -pub struct NetworkSyscall; +pub struct NetworkSyscall { + client: reqwest::Client, +} impl NetworkSyscall { pub fn new() -> Self { - Self + let client = reqwest::Client::builder() + .connect_timeout(Duration::from_secs(5)) + .read_timeout(Duration::from_secs(30)) + .timeout(Duration::from_secs(60)) + .redirect(reqwest::redirect::Policy::limited(5)) + .build() + .unwrap_or_else(|_| reqwest::Client::new()); + Self { client } } - pub fn available(&self) -> bool { - // Simple connectivity check: try to resolve a known host - std::net::TcpStream::connect_timeout( - &std::net::SocketAddr::from(([8, 8, 8, 8], 53)), - std::time::Duration::from_secs(2), + pub async fn available(&self) -> bool { + tokio::time::timeout( + Duration::from_secs(2), + tokio::net::TcpStream::connect(SocketAddr::from(([8, 8, 8, 8], 53))), ) - .is_ok() + .await + .map(|result| result.is_ok()) + .unwrap_or(false) } pub async fn request( &self, url: &str, method: &str, - headers: &std::collections::HashMap, + headers: &HashMap, body: &[u8], - ) -> Result<(i32, std::collections::HashMap, Vec), String> { - // Minimal HTTP client using tokio TcpStream + manual HTTP. - // For production, use hyper or reqwest. For now, shell out to a basic impl. - use tokio::process::Command; + ) -> Result<(i32, HashMap, Vec), String> { + let method = method + .parse::() + .map_err(|e| format!("invalid HTTP method {method:?}: {e}"))?; + let mut request = self.client.request(method, url); - let mut cmd = Command::new("curl"); - cmd.arg("-s") - .arg("-X") - .arg(method) - .arg("-o") - .arg("-") - .arg("-w") - .arg("\n%{http_code}"); - - for (k, v) in headers { - cmd.arg("-H").arg(format!("{}: {}", k, v)); + for (name, value) in headers { + let header_name = name + .parse::() + .map_err(|e| format!("invalid HTTP header name {name:?}: {e}"))?; + let header_value = value + .parse::() + .map_err(|e| format!("invalid HTTP header value for {name}: {e}"))?; + request = request.header(header_name, header_value); } if !body.is_empty() { - cmd.arg("-d").arg(String::from_utf8_lossy(body).to_string()); + request = request.body(body.to_vec()); } - cmd.arg(url); + let response = request.send().await.map_err(|e| e.to_string())?; - let output = cmd.output().await.map_err(|e| e.to_string())?; - let raw = String::from_utf8_lossy(&output.stdout); - let lines: Vec<&str> = raw.rsplitn(2, '\n').collect(); + // Check content-length before downloading body + if let Some(len) = response.content_length() { + if len > MAX_RESPONSE_BODY { + return Err(format!( + "response body too large: {} bytes (limit {} bytes)", + len, MAX_RESPONSE_BODY + )); + } + } - let status = lines - .first() - .and_then(|s| s.parse::().ok()) - .unwrap_or(0); - let response_body = lines.get(1).unwrap_or(&"").as_bytes().to_vec(); + let status = i32::from(response.status().as_u16()); + let resp_headers = response + .headers() + .iter() + .map(|(name, value)| { + ( + name.to_string(), + value.to_str().unwrap_or_default().to_string(), + ) + }) + .collect(); + let resp_body = response.bytes().await.map_err(|e| e.to_string())?; + if resp_body.len() as u64 > MAX_RESPONSE_BODY { + return Err(format!( + "response body too large: {} bytes (limit {} bytes)", + resp_body.len(), + MAX_RESPONSE_BODY + )); + } - Ok((status, std::collections::HashMap::new(), response_body)) + Ok((status, resp_headers, resp_body.to_vec())) } } diff --git a/core/agentsd/src/syscall/process.rs b/core/agentsd/src/syscall/process.rs index dd88f25..6366c58 100644 --- a/core/agentsd/src/syscall/process.rs +++ b/core/agentsd/src/syscall/process.rs @@ -1,23 +1,31 @@ use std::collections::HashMap; -use std::sync::Mutex; +use std::sync::Arc; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::process::Command; +use tokio::sync::{mpsc, Mutex, RwLock}; +/// Per-process managed state, decoupled from the Child handle. struct ManagedProcess { cmd: String, + status: Arc>, + stdin_tx: Option>>, + stdout_rx: Arc>>>, + /// Notified when the process exits. Clone the receiver to wait. + exit_notify: Arc, + /// Handle to abort background tasks on kill. + _tasks: Vec>, } /// Process syscall: spawn and manage subprocesses. /// Uses the real OS PID as the identifier. pub struct ProcessSyscall { - procs: Mutex>, - children: tokio::sync::Mutex>, + procs: RwLock>>, } impl ProcessSyscall { pub fn new() -> Self { Self { - procs: Mutex::new(HashMap::new()), - children: tokio::sync::Mutex::new(HashMap::new()), + procs: RwLock::new(HashMap::new()), } } @@ -39,86 +47,244 @@ impl ProcessSyscall { command .stdin(std::process::Stdio::piped()) .stdout(std::process::Stdio::piped()) - .stderr(std::process::Stdio::piped()); + .stderr(std::process::Stdio::piped()) + .kill_on_drop(true); - let child = command.spawn().map_err(|e| e.to_string())?; + let mut child = command.spawn().map_err(|e| e.to_string())?; let pid = child .id() .ok_or_else(|| "spawned process has no OS pid".to_string())? as u64; - self.procs.lock().unwrap().insert( - pid, - ManagedProcess { - cmd: cmd.to_string(), - }, - ); - self.children.lock().await.insert(pid, child); + // Take ownership of stdio handles + let child_stdin = child.stdin.take(); + let child_stdout = child.stdout.take(); + let child_stderr = child.stderr.take(); + let status = Arc::new(Mutex::new("running".to_string())); + let exit_notify = Arc::new(tokio::sync::Notify::new()); + + // Stdin writer task + let (stdin_tx, mut stdin_rx) = mpsc::channel::>(64); + let stdin_task = tokio::spawn(async move { + if let Some(mut stdin) = child_stdin { + while let Some(data) = stdin_rx.recv().await { + if stdin.write_all(&data).await.is_err() { + break; + } + if stdin.flush().await.is_err() { + break; + } + } + } + }); + + // Stdout reader task + let (stdout_tx, stdout_rx) = mpsc::channel::>(64); + let stdout_task = tokio::spawn(async move { + if let Some(mut stdout) = child_stdout { + loop { + let mut buf = vec![0u8; 4096]; + match stdout.read(&mut buf).await { + Ok(0) => break, + Ok(n) => { + buf.truncate(n); + if stdout_tx.send(buf).await.is_err() { + break; + } + } + Err(_) => break, + } + } + } + }); + + // Stderr drain task (prevent pipe fill-up) + let stderr_task = tokio::spawn(async move { + if let Some(mut stderr) = child_stderr { + let mut buf = vec![0u8; 4096]; + loop { + match stderr.read(&mut buf).await { + Ok(0) | Err(_) => break, + Ok(_) => {} // discard + } + } + } + }); + + // Reaper task: waits for exit and updates status + let status_clone = status.clone(); + let exit_notify_clone = exit_notify.clone(); + let reaper_task = tokio::spawn(async move { + let exit_status = child.wait().await; + let code = exit_status.map(|s| s.code().unwrap_or(-1)).unwrap_or(-1); + *status_clone.lock().await = format!("exited({})", code); + exit_notify_clone.notify_waiters(); + }); + + let managed = Arc::new(ManagedProcess { + cmd: cmd.to_string(), + status, + stdin_tx: Some(stdin_tx), + stdout_rx: Arc::new(Mutex::new(stdout_rx)), + exit_notify, + _tasks: vec![stdin_task, stdout_task, stderr_task, reaper_task], + }); + + self.procs.write().await.insert(pid, managed); tracing::info!("process spawned: pid={} cmd={}", pid, cmd); Ok(pid) } pub async fn kill(&self, pid: u64) -> Result<(), String> { - let mut child = { - let mut children = self.children.lock().await; - children.remove(&pid).ok_or_else(|| format!("pid {} not found", pid))? - }; + let procs = self.procs.read().await; + let proc = procs + .get(&pid) + .ok_or_else(|| format!("pid {} not found", pid))? + .clone(); + drop(procs); - child.kill().await.map_err(|e| e.to_string())?; - self.procs.lock().unwrap().remove(&pid); + // Abort all background tasks (stdin writer, stdout reader, stderr drain, reaper) + for task in proc._tasks.iter() { + task.abort(); + } + *proc.status.lock().await = "killed".to_string(); + proc.exit_notify.notify_waiters(); + + // Also send OS kill + #[cfg(unix)] + { + let os_pid = pid as i32; + unsafe { libc::kill(os_pid as libc::pid_t, 9) }; + } + #[cfg(not(unix))] + { + // On Windows, aborting the reaper task triggers kill_on_drop on the Child + } + + self.procs.write().await.remove(&pid); tracing::info!("process killed: pid={}", pid); Ok(()) } pub async fn wait(&self, pid: u64) -> Result { - let mut child = { - let mut children = self.children.lock().await; - children.remove(&pid).ok_or_else(|| format!("pid {} not found", pid))? - }; - self.procs.lock().unwrap().remove(&pid); - let status = child.wait().await.map_err(|e| e.to_string())?; - Ok(status.code().unwrap_or(-1)) + let procs = self.procs.read().await; + let proc = procs + .get(&pid) + .ok_or_else(|| format!("pid {} not found", pid))? + .clone(); + drop(procs); + + // Check if already exited before waiting + { + let status = proc.status.lock().await; + if status.starts_with("exited") || *status == "killed" { + let s = status.clone(); + drop(status); + self.procs.write().await.remove(&pid); + return parse_exit_code(&s); + } + } + + // Wait for exit notification + proc.exit_notify.notified().await; + let status = proc.status.lock().await.clone(); + + // Clean up from map + self.procs.write().await.remove(&pid); + parse_exit_code(&status) } pub async fn write_stdin(&self, pid: u64, data: &[u8]) -> Result<(), String> { - use tokio::io::AsyncWriteExt; - let mut children = self.children.lock().await; - if let Some(child) = children.get_mut(&pid) { - if let Some(stdin) = child.stdin.as_mut() { - stdin.write_all(data).await.map_err(|e| e.to_string())?; - stdin.flush().await.map_err(|e| e.to_string())?; - Ok(()) - } else { - Err("stdin not available".to_string()) - } + let procs = self.procs.read().await; + let proc = procs + .get(&pid) + .ok_or_else(|| format!("pid {} not found", pid))? + .clone(); + drop(procs); + + if let Some(ref tx) = proc.stdin_tx { + tx.send(data.to_vec()) + .await + .map_err(|_| "stdin channel closed".to_string()) } else { - Err(format!("pid {} not found", pid)) + Err("stdin not available".to_string()) } } pub async fn read_stdout(&self, pid: u64, buf: &mut [u8]) -> Result { - use tokio::io::AsyncReadExt; - let mut children = self.children.lock().await; - if let Some(child) = children.get_mut(&pid) { - if let Some(stdout) = child.stdout.as_mut() { - stdout.read(buf).await.map_err(|e| e.to_string()) - } else { - Err("stdout not available".to_string()) + let procs = self.procs.read().await; + let proc = procs + .get(&pid) + .ok_or_else(|| format!("pid {} not found", pid))? + .clone(); + drop(procs); + + let mut rx = proc.stdout_rx.lock().await; + match rx.recv().await { + Some(data) => { + let n = std::cmp::min(data.len(), buf.len()); + buf[..n].copy_from_slice(&data[..n]); + Ok(n) } - } else { - Err(format!("pid {} not found", pid)) + None => Ok(0), // EOF } } - pub async fn signal(&self, pid: u64, _signal: i32) -> Result<(), String> { - self.kill(pid).await + pub async fn signal(&self, pid: u64, signal: i32) -> Result<(), String> { + if signal == 9 { + return self.kill(pid).await; + } + + let procs = self.procs.read().await; + if !procs.contains_key(&pid) { + return Err(format!("pid {} not found", pid)); + } + drop(procs); + + #[cfg(unix)] + { + let os_pid = i32::try_from(pid) + .map_err(|_| format!("pid {} out of range for platform signal API", pid))?; + let result = unsafe { libc::kill(os_pid as libc::pid_t, signal as libc::c_int) }; + if result == 0 { + Ok(()) + } else { + Err(format!( + "signal {} to pid {} failed: {}", + signal, + pid, + std::io::Error::last_os_error() + )) + } + } + + #[cfg(not(unix))] + { + Err(format!( + "process.signal only supports signal 9 on this platform, got {}", + signal + )) + } } - pub fn list(&self) -> Vec<(u64, String, String)> { - let procs = self.procs.lock().unwrap(); - procs - .iter() - .map(|(pid, mp)| (*pid, mp.cmd.clone(), "running".to_string())) - .collect() + pub async fn list(&self) -> Vec<(u64, String, String)> { + let procs = self.procs.read().await; + let mut result = Vec::with_capacity(procs.len()); + for (pid, mp) in procs.iter() { + let status = mp.status.lock().await.clone(); + result.push((*pid, mp.cmd.clone(), status)); + } + result + } +} + +fn parse_exit_code(status: &str) -> Result { + if let Some(code_str) = status + .strip_prefix("exited(") + .and_then(|s| s.strip_suffix(')')) + { + code_str.parse::().map_err(|e| e.to_string()) + } else { + Ok(-1) } } diff --git a/core/agentsd/src/syscall/timer.rs b/core/agentsd/src/syscall/timer.rs index 033e8b4..49663d8 100644 --- a/core/agentsd/src/syscall/timer.rs +++ b/core/agentsd/src/syscall/timer.rs @@ -52,12 +52,10 @@ impl TimerSyscall { timers.lock().unwrap().remove(&tid); }); - self.timers.lock().unwrap().insert( - timer_id.clone(), - TimerEntry { - handle, - }, - ); + self.timers + .lock() + .unwrap() + .insert(timer_id.clone(), TimerEntry { handle }); timer_id } diff --git a/plugins/ai_test/src/main.rs b/plugins/ai_test/src/main.rs index 066e5dc..3d73ca8 100644 --- a/plugins/ai_test/src/main.rs +++ b/plugins/ai_test/src/main.rs @@ -304,7 +304,7 @@ impl CheckRunner { .into_inner(); let names: Vec<&str> = list.entries.iter().map(|e| e.name.as_str()).collect(); ensure!( - list.ok && names.iter().any(|name| *name == "sample.txt"), + list.ok && names.contains(&"sample.txt"), "list mismatch ok={} names={:?} error={:?}", list.ok, names, @@ -341,12 +341,13 @@ impl CheckRunner { async fn process_roundtrip(&mut self) -> Result { let mut process = ProcessClient::new(self.channel.clone()); + let (cmd, args) = process_command(); let resp = process .spawn(request_with_plugin( PLUGIN_ID, SpawnRequest { - cmd: "cmd".to_string(), - args: vec!["/C".to_string(), format!("echo {PROCESS_OK}")], + cmd, + args, env: HashMap::new(), cwd: executable_dir() .map(|path| path_string(&path)) @@ -504,6 +505,20 @@ fn executable_data_dir() -> PathBuf { .join("data") } +fn process_command() -> (String, Vec) { + if cfg!(windows) { + ( + "cmd".to_string(), + vec!["/C".to_string(), format!("echo {PROCESS_OK}")], + ) + } else { + ( + "sh".to_string(), + vec!["-c".to_string(), format!("printf %s {PROCESS_OK}")], + ) + } +} + fn path_string(path: &std::path::Path) -> String { path.to_string_lossy().to_string() } diff --git a/plugins/claudecode/Cargo.toml b/plugins/claudecode/Cargo.toml index e926ca9..bef1bff 100644 --- a/plugins/claudecode/Cargo.toml +++ b/plugins/claudecode/Cargo.toml @@ -14,6 +14,8 @@ anyhow = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true } +tokio-stream = { workspace = true } tonic = { workspace = true } futures-core = { workspace = true } tracing = { workspace = true } +tracing-subscriber = { workspace = true } diff --git a/plugins/claudecode/src/main.rs b/plugins/claudecode/src/main.rs index 81455fa..d33d5bc 100644 --- a/plugins/claudecode/src/main.rs +++ b/plugins/claudecode/src/main.rs @@ -12,7 +12,7 @@ use tonic::{transport::Server, Request, Response, Status}; const PLUGIN_ID: &str = "claudecode"; const DEFAULT_PORT: u16 = 50053; -#[derive(Debug, Deserialize)] +#[derive(Debug, Default, Deserialize)] struct CallArgs { #[serde(default)] prompt: String, @@ -32,26 +32,22 @@ fn find_claude_bin() -> Option { return Some(bin); } - let candidates = vec![ - "claude", - "claude.cmd", - &std::env::var("APPDATA") - .map(|p| format!("{}\\npm\\claude.cmd", p)) - .unwrap_or_default(), - ]; + let appdata_claude = std::env::var("APPDATA") + .ok() + .map(|p| format!("{}\\npm\\claude.cmd", p)); + let mut candidates = vec!["claude".to_string(), "claude.cmd".to_string()]; + if let Some(path) = appdata_claude { + candidates.push(path); + } - for candidate in candidates { - if std::process::Command::new(candidate) + candidates.into_iter().find(|candidate| { + std::process::Command::new(candidate) .arg("--version") .stdout(Stdio::null()) .stderr(Stdio::null()) .status() .is_ok() - { - return Some(candidate.to_string()); - } - } - None + }) } async fn call_claude( @@ -97,9 +93,12 @@ async fn call_claude( drop(stdin); } - let output = tokio::time::timeout(std::time::Duration::from_secs(300), child.wait_with_output()) - .await - .context("claude timeout (300s)")??; + let output = tokio::time::timeout( + std::time::Duration::from_secs(300), + child.wait_with_output(), + ) + .await + .context("claude timeout (300s)")??; if !output.status.success() { let stderr = String::from_utf8_lossy(&output.stderr); @@ -148,8 +147,10 @@ impl ClaudeCodePlugin { .await } "review" => { - let review_prompt = - format!("Review the following code or changes. Be concise and actionable:\n\n{}", prompt); + let review_prompt = format!( + "Review the following code or changes. Be concise and actionable:\n\n{}", + prompt + ); call_claude( &self.claude_bin, &review_prompt, @@ -184,7 +185,10 @@ impl ClaudeCodePlugin { #[tonic::async_trait] impl PluginBridge for ClaudeCodePlugin { - async fn register(&self, _req: Request) -> Result, Status> { + async fn register( + &self, + _req: Request, + ) -> Result, Status> { Err(Status::unimplemented("not a bridge")) } @@ -195,13 +199,19 @@ impl PluginBridge for ClaudeCodePlugin { Ok(Response::new(response)) } - type StreamCallStream = futures_core::stream::Empty>; + type StreamCallStream = tokio_stream::Empty>; - async fn stream_call(&self, _req: Request) -> Result, Status> { + async fn stream_call( + &self, + _req: Request, + ) -> Result, Status> { Err(Status::unimplemented("stream_call not supported")) } - async fn heartbeat(&self, _req: Request) -> Result, Status> { + async fn heartbeat( + &self, + _req: Request, + ) -> Result, Status> { Ok(Response::new(HeartbeatResponse { ok: true })) } } @@ -210,7 +220,8 @@ impl PluginBridge for ClaudeCodePlugin { async fn main() -> Result<()> { tracing_subscriber::fmt::init(); - let claude_bin = find_claude_bin().context("Claude Code CLI not found. Install: npm install -g @anthropic-ai/claude-code")?; + let claude_bin = find_claude_bin() + .context("Claude Code CLI not found. Install: npm install -g @anthropic-ai/claude-code")?; tracing::info!("Found Claude Code CLI: {}", claude_bin); let endpoint = agentsd_plugin_sdk::plugin_endpoint_from_env(DEFAULT_PORT); @@ -224,21 +235,42 @@ async fn main() -> Result<()> { "Claude Code", env!("CARGO_PKG_VERSION"), "standard", - &["process.spawn", "display.text", "fs.read", "fs.write", "network.request"], + &[ + "process.spawn", + "display.text", + "fs.read", + "fs.write", + "network.request", + ], &["hermes"], vec![ - export("chat", "Send prompt to Claude Code, get text response (no tools)"), - export("code", "Generate or edit code with Claude Code (Bash/Read/Edit/Write)"), + export( + "chat", + "Send prompt to Claude Code, get text response (no tools)", + ), + export( + "code", + "Generate or edit code with Claude Code (Bash/Read/Edit/Write)", + ), export("review", "Review code changes (Read/Glob/Grep only)"), - export("ask", "Ask about the codebase (Read/Glob/Grep, max 3 turns)"), - export("run", "Full Claude Code access (all tools, unlimited turns)"), + export( + "ask", + "Ask about the codebase (Read/Glob/Grep, max 3 turns)", + ), + export( + "run", + "Full Claude Code access (all tools, unlimited turns)", + ), ], &endpoint, ), ) .await?; - tracing::info!("Registered with agentsd: session={}", register_resp.session_id); + tracing::info!( + "Registered with agentsd: session={}", + register_resp.session_id + ); let plugin = ClaudeCodePlugin::new(claude_bin); let addr = listen_addr.parse()?; diff --git a/plugins/echo/src/main.rs b/plugins/echo/src/main.rs index 1680f0e..918f30c 100644 --- a/plugins/echo/src/main.rs +++ b/plugins/echo/src/main.rs @@ -26,7 +26,10 @@ async fn main() -> Result<()> { .await? .into_inner(); ensure!(register.ok, "register failed: {}", register.error); - println!("Register: ok={} session={}", register.ok, register.session_id); + println!( + "Register: ok={} session={}", + register.ok, register.session_id + ); let mut display = DisplayClient::new(channel.clone()); let resp = display @@ -84,7 +87,11 @@ async fn main() -> Result<()> { ensure!(resp.ok, "FS.List failed: {}", resp.error); println!("FS.List: ok={} entries={}", resp.ok, resp.entries.len()); for entry in resp.entries.iter().take(5) { - println!(" {} {}", entry.name, if entry.is_dir { "[dir]" } else { "" }); + println!( + " {} {}", + entry.name, + if entry.is_dir { "[dir]" } else { "" } + ); } println!("\nAll syscalls verified OK!"); diff --git a/plugins/hermes/src/main.rs b/plugins/hermes/src/main.rs index 4a3a071..62516c7 100644 --- a/plugins/hermes/src/main.rs +++ b/plugins/hermes/src/main.rs @@ -44,7 +44,9 @@ impl PluginBridge for HermesPlugin { &self, _req: Request, ) -> std::result::Result, Status> { - Err(Status::unimplemented("hermes plugin is not an agentsd bridge")) + Err(Status::unimplemented( + "hermes plugin is not an agentsd bridge", + )) } async fn call( @@ -62,13 +64,16 @@ impl PluginBridge for HermesPlugin { Ok(Response::new(response)) } - type StreamCallStream = tokio_stream::wrappers::ReceiverStream>; + type StreamCallStream = + tokio_stream::wrappers::ReceiverStream>; async fn stream_call( &self, _req: Request, ) -> std::result::Result, Status> { - Err(Status::unimplemented("stream_call is not implemented for hermes")) + Err(Status::unimplemented( + "stream_call is not implemented for hermes", + )) } async fn heartbeat( @@ -110,7 +115,7 @@ async fn main() -> Result<()> { println!("Hermes PluginBridge server listening on {}", listen_addr); Server::builder() - .add_service(PluginBridgeServer::new(HermesPlugin::default())) + .add_service(PluginBridgeServer::new(HermesPlugin)) .serve_with_incoming(incoming) .await?; Ok(()) diff --git a/plugins/plugin-sdk/src/lib.rs b/plugins/plugin-sdk/src/lib.rs index 729c911..ac3c611 100644 --- a/plugins/plugin-sdk/src/lib.rs +++ b/plugins/plugin-sdk/src/lib.rs @@ -59,6 +59,7 @@ pub fn plugin_info( } } +#[allow(clippy::too_many_arguments)] pub fn plugin_info_with_endpoint( id: &str, name: &str, @@ -95,7 +96,11 @@ pub fn encode_json(value: &T) -> Result> { serde_json::to_vec(value).context("encode JSON") } -pub fn call_response_json(value: &T, ok: bool, error: impl Into) -> CallResponse { +pub fn call_response_json( + value: &T, + ok: bool, + error: impl Into, +) -> CallResponse { match encode_json(value) { Ok(result) => CallResponse { result,