diff --git a/README.md b/README.md index a6ffc31..260e00c 100644 --- a/README.md +++ b/README.md @@ -37,6 +37,7 @@ Codex CLI（0.160.0 以上）をアカウント種別から選べます。対話 - トピックの再開。一覧から選ぶとそのトピックが部屋へ戻り、アカウントに `再開コマンド` があればそこに居た CLI セッションも戻ります - 参加者が自分でトピックの過去発言を引く道具（サイドカーの `read_room_history`） - 席が自分の利用上限（5 時間と週の使用率とリセット時刻、アプリが受け取った時刻）を読む道具（サイドカーの `my_usage`） +- 参加 AI が自分や他の席を開始・終了・会話を保って再起動する [MCP セッション操作](docs/1-room.md#ai-からのセッション操作415) - 設定メニュー。アプリ本体が起動するローカル MCP サーバ（同梱の `github-webhook-mcp` ブリッジなど）の一覧・実行状態・コマンドと引数と環境変数の編集・ログを扱い、設定はアプリのデータディレクトリの `mcp-servers.json` に保存します - Windows 上で `npm ci` と Rust のコンパイル確認を行う CI - GitHub Release 公開時に Tauri バンドルを作成する CD diff --git a/crates/mcp-config/Cargo.lock b/crates/mcp-config/Cargo.lock index ad9d816..456bb54 100644 --- a/crates/mcp-config/Cargo.lock +++ b/crates/mcp-config/Cargo.lock @@ -114,6 +114,7 @@ name = "mcp-config" version = "0.1.0" dependencies = [ "chrono", + "serde", "serde_json", "toml_edit", "uuid", @@ -183,6 +184,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" dependencies = [ "serde_core", + "serde_derive", ] [[package]] diff --git a/crates/mcp-config/Cargo.toml b/crates/mcp-config/Cargo.toml index 51ce3fa..9a70bb3 100644 --- a/crates/mcp-config/Cargo.toml +++ b/crates/mcp-config/Cargo.toml @@ -12,6 +12,7 @@ edition = "2021" # GNU target (see the CI note in docs/5-development.md). [dependencies] +serde = { version = "1", features = ["derive"] } serde_json = "1" chrono = { version = "0.4", default-features = false, features = ["std"] } toml_edit = "0.20.2" diff --git a/crates/mcp-config/src/codex.rs b/crates/mcp-config/src/codex.rs index 91e777a..b9327cb 100644 --- a/crates/mcp-config/src/codex.rs +++ b/crates/mcp-config/src/codex.rs @@ -15,6 +15,7 @@ pub const NATIVE_URL_ENV: &str = "PULLCEPT_NATIVE_URL"; pub const OUTPUT_STYLE_ENV: &str = "LI_PLUS_OUTPUT_STYLE"; pub const NATIVE_PATH: &str = "/hooks/codex-session"; pub const SIDECAR_ENV: &[&str] = &[ + crate::session_control::CAPABILITY_ENV, LAUNCH_ID_ENV, admission::REQUIRED_ENV, ROOM_TOKEN_ENV, @@ -54,7 +55,7 @@ pub(crate) fn room_server_definition(own: &str, runner: &str, entry: &str) -> St .join(","); // The only registration (#297). The whole table, so it works before project trust // and an entry left in a trusted project's config cannot override this launch. - format!("mcp_servers.{own}={{command='node',args=[{},{}],env_vars=[{vars}],enabled=true,tools={{say_to_room={{approval_mode='approve'}},read_room_history={{approval_mode='approve'}},my_usage={{approval_mode='approve'}}}}}}", instruction_value(runner), instruction_value(entry)) + format!("mcp_servers.{own}={{command='node',args=[{},{}],env_vars=[{vars}],enabled=true,tools={{say_to_room={{approval_mode='approve'}},read_room_history={{approval_mode='approve'}},my_usage={{approval_mode='approve'}},list_account_sessions={{approval_mode='approve'}},account_session_status={{approval_mode='approve'}},control_account_session={{approval_mode='approve'}}}}}}", instruction_value(runner), instruction_value(entry)) } /// The runner and entry paths as the room server's table carries them, or why diff --git a/crates/mcp-config/src/lib.rs b/crates/mcp-config/src/lib.rs index 213321a..4c2b661 100644 --- a/crates/mcp-config/src/lib.rs +++ b/crates/mcp-config/src/lib.rs @@ -41,6 +41,7 @@ pub mod claude_resume; pub mod permission_prompt; pub mod prompt_hold; pub mod usage; +pub mod session_control; /// Prefix of the name the sidecar is registered under in `.mcp.json`. /// @@ -801,6 +802,7 @@ pub const ROOM_ID_ENV: &str = "PULLCEPT_ROOM_ID"; /// when the CLI is spawned, so the app's value holds even if that refusal /// were bypassed. pub const APP_LAUNCH_ENV: &[&str] = &[ + session_control::CAPABILITY_ENV, ROOM_TOKEN_ENV, LAUNCHED_AS_ENV, ROOM_ID_ENV, diff --git a/crates/mcp-config/src/session_control.rs b/crates/mcp-config/src/session_control.rs new file mode 100644 index 0000000..67a3936 --- /dev/null +++ b/crates/mcp-config/src/session_control.rs @@ -0,0 +1,527 @@ +//! App-owned, retry-safe session operation records (#415). +use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; + +pub const CAPABILITY_ENV: &str = "PULLCEPT_CONTROL_CAPABILITY"; +pub struct ControlBinding { + pub topic: String, + pub account: String, + pub capability: String, +} +impl ControlBinding { + pub fn matches_hello(&self, topic: Option<&str>, account: Option<&str>) -> bool { + topic == Some(self.topic.as_str()) && account == Some(self.account.as_str()) + } +} +pub fn capability_matches( + supplied: Option<&str>, + expected: Option<&str>, + live_or_starting: bool, +) -> bool { + live_or_starting && supplied.is_some_and(|token| !token.is_empty() && expected == Some(token)) +} + +#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct Request { + pub operation_id: String, + pub action: Action, + pub topic_id: String, + pub account_id: String, + pub expected_pty_id: Option, + pub mode: Option, +} + +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Action { + Start, + Stop, + Restart, +} +#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Mode { + Fresh, + Resume, +} + +#[derive(Clone)] +pub struct Connection { + pub socket: String, + pub topic: String, + pub account: String, + pub generation: Option, + pub launch: Option, +} +impl Connection { + pub fn matches_live(&self, live: bool, generation: Option<&str>, launch: Option<&str>) -> bool { + live && match &self.launch { + Some(expected) => launch == Some(expected.as_str()), + None => self + .generation + .as_deref() + .is_some_and(|expected| generation == Some(expected)), + } + } + pub fn authorize( + &self, + socket: &str, + live: bool, + generation: Option<&str>, + launch: Option<&str>, + ) -> Result<(), String> { + if self.socket != socket || !self.matches_live(live, generation, launch) { + return Err("Requester's socket or session changed".into()); + } + Ok(()) + } +} + +pub struct Current { + pub pty_id: String, + pub native_id: Option, + pub resumed_from: Option, +} +/// The app holds its lifecycle lock for this whole sequence. Drivers resolve +/// the exact configured account/topic; they never accept a command from MCP. +pub trait Driver { + fn current(&mut self, request: &Request) -> Result, String>; + fn resume(&mut self, request: &Request) -> Result; + fn stop(&mut self, request: &Request) -> Result<(), String>; + fn start( + &mut self, + request: &Request, + mode: Mode, + resume_id: Option<&str>, + ) -> Result; +} + +pub fn execute(request: &Request, driver: &mut impl Driver) -> Result, String> { + request.validate()?; + let current = driver.current(request)?; + request.check_target(current.as_ref().map(|c| c.pty_id.as_str()))?; + let mode = if request.action == Action::Restart { + Mode::Resume + } else { + request.mode.unwrap_or(Mode::Fresh) + }; + let resume = if mode == Mode::Resume { + let id = driver.resume(request)?; + if request.action == Action::Restart + && current.as_ref().is_none_or(|c| { + c.native_id.as_deref() != Some(&id) && c.resumed_from.as_deref() != Some(&id) + }) + { + return Err( + "Current session has no matching captured resume ID yet; it was left running" + .into(), + ); + } + Some(id) + } else { + None + }; + if matches!(request.action, Action::Stop | Action::Restart) { + driver.stop(request)?; + } + if request.action == Action::Stop { + return Ok(None); + } + driver.start(request, mode, resume.as_deref()).map(Some) +} + +impl Request { + pub fn validate(&self) -> Result<(), String> { + for id in [&self.operation_id, &self.topic_id, &self.account_id] { + if id.trim().is_empty() || id.len() > 200 { + return Err("IDs must contain 1..200 bytes".into()); + } + } + match self.action { + Action::Start if self.expected_pty_id.is_none() && self.mode.is_some() => Ok(()), + Action::Stop | Action::Restart if self.expected_pty_id.as_deref().is_some_and(|id| !id.is_empty()) && self.mode.is_none() => Ok(()), + _ => Err("start requires explicit fresh/resume mode; stop/restart require expected_pty_id and no mode".into()), + } + } + pub fn check_target(&self, current: Option<&str>) -> Result<(), String> { + match self.action { + Action::Start if current.is_none() => Ok(()), + Action::Stop | Action::Restart + if current.is_some() && current == self.expected_pty_id.as_deref() => + { + Ok(()) + } + _ => Err("target seat changed, is starting, or is not in the required state".into()), + } + } +} + +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct Operation { + pub request: Request, + pub requester_account: String, + pub requester_topic: String, + pub requester_name: String, + pub status: String, + pub accepted_at: String, + pub updated_at: String, + pub new_pty_id: Option, + pub error: Option, +} + +#[derive(Clone, Default, Deserialize, Serialize)] +pub struct Book { + pub operations: BTreeMap, +} +impl Book { + pub fn accept_durable( + &mut self, + request: Request, + account: &str, + topic: &str, + name: &str, + at: &str, + persist: impl FnOnce(&Book) -> Result<(), String>, + ) -> Result { + let previous = self.clone(); + let accepted = self.accept(request, account, topic, name, at)?; + if accepted { + if let Err(error) = persist(self) { + *self = previous; + return Err(error); + } + } + Ok(accepted) + } + /// Return false for an identical retry. IDs cannot be reused by another caller. + pub fn accept( + &mut self, + request: Request, + account: &str, + topic: &str, + name: &str, + at: &str, + ) -> Result { + request.validate()?; + if let Some(old) = self.operations.get(&request.operation_id) { + return if old.request == request + && old.requester_account == account + && old.requester_topic == topic + { + Ok(false) + } else { + Err("operation_id already belongs to a different request or requester".into()) + }; + } + // Keep requests until explicit app-data removal: forgetting an ID permits replay. + self.operations.insert( + request.operation_id.clone(), + Operation { + request, + requester_account: account.into(), + requester_topic: topic.into(), + requester_name: name.into(), + status: "accepted".into(), + accepted_at: at.into(), + updated_at: at.into(), + new_pty_id: None, + error: None, + }, + ); + Ok(true) + } + pub fn interrupt(&mut self, at: &str) { + for op in self + .operations + .values_mut() + .filter(|op| matches!(op.status.as_str(), "accepted" | "running")) + { + op.status = "interrupted".into(); + op.updated_at = at.into(); + op.error = Some("Pullcept exited before completion; operation was not replayed".into()); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + fn request(action: Action) -> Request { + Request { + operation_id: "op".into(), + action, + topic_id: "room".into(), + account_id: "target".into(), + expected_pty_id: (action != Action::Start).then(|| "old".into()), + mode: (action == Action::Start).then_some(Mode::Fresh), + } + } + #[test] + fn caller_authority_is_live_launch_and_current_socket_bound() { + // A bearer-token client cannot gain control by naming a live account. + assert!(!capability_matches(None, Some("secret"), true)); + assert!(!capability_matches( + Some("other-secret"), + Some("secret"), + true + )); + assert!(!capability_matches( + Some("old-secret"), + Some("new-secret"), + true + )); + assert!(!capability_matches(Some("secret"), Some("secret"), false)); + assert!(capability_matches(Some("secret"), Some("secret"), true)); + let mut caller = Connection { + socket: "socket".into(), + topic: "room".into(), + account: "caller".into(), + generation: Some("old".into()), + launch: None, + }; + assert!(caller.authorize("socket", true, Some("old"), None).is_ok()); + assert!(caller + .authorize("replaced-socket", true, Some("old"), None) + .is_err()); + assert!(caller + .authorize("socket", false, Some("old"), None) + .is_err()); + assert!(caller.authorize("socket", true, Some("new"), None).is_err()); + assert!(caller.authorize("socket", true, None, None).is_err()); + caller.launch = Some("launch".into()); + assert!(caller + .authorize("socket", true, Some("old"), Some("launch")) + .is_ok()); + assert!(caller + .authorize("socket", true, Some("old"), Some("other-launch")) + .is_err()); + } + + #[test] + fn hello_cannot_change_capability_scope_and_early_connection_survives_hold() { + let binding = ControlBinding { + topic: "topic".into(), + account: "account".into(), + capability: "capability".into(), + }; + assert!(binding.matches_hello(Some("topic"), Some("account"))); + assert!(!binding.matches_hello(Some("other-topic"), Some("account"))); + assert!(!binding.matches_hello(Some("topic"), Some("other-account"))); + assert!(!binding.matches_hello(None, Some("account"))); + let connection = Connection { + socket: "socket".into(), + topic: binding.topic, + account: binding.account, + generation: Some(binding.capability), + launch: None, + }; + // Captured during Starting: no PTY is part of this connection record. + assert!(connection + .authorize("socket", false, Some("capability"), None) + .is_err()); + assert!(connection + .authorize("socket", true, Some("capability"), None) + .is_ok()); + assert!(connection + .authorize("socket", true, Some("new-capability"), None) + .is_err()); + } + + #[test] + fn concurrent_identical_requests_are_accepted_once() { + let book = std::sync::Arc::new(std::sync::Mutex::new(Book::default())); + let threads: Vec<_> = (0..12) + .map(|_| { + let book = book.clone(); + std::thread::spawn(move || { + book.lock() + .unwrap() + .accept(request(Action::Start), "caller", "room", "Caller", "now") + .unwrap() + }) + }) + .collect(); + let accepted = threads + .into_iter() + .map(|worker| usize::from(worker.join().unwrap())) + .sum::(); + assert_eq!(accepted, 1); + assert_eq!(book.lock().unwrap().operations.len(), 1); + } + + #[test] + fn failed_journal_write_does_not_accept_or_schedule_an_operation() { + let mut book = Book::default(); + assert!(book + .accept_durable( + request(Action::Restart), + "caller", + "room", + "Caller", + "now", + |_| Err("disk unavailable".into()) + ) + .is_err()); + assert!(book.operations.is_empty()); + assert!(book + .accept_durable( + request(Action::Restart), + "caller", + "room", + "Caller", + "later", + |_| Ok(()) + ) + .unwrap()); + assert!(!book + .accept_durable( + request(Action::Restart), + "caller", + "room", + "Caller", + "later", + |_| panic!("identical retry must not write or schedule") + ) + .unwrap()); + } + #[test] + fn retries_and_requester_identity() { + let mut book = Book::default(); + let req = request(Action::Restart); + assert!(book + .accept(req.clone(), "caller", "room", "Caller", "now") + .unwrap()); + assert!(!book + .accept(req.clone(), "caller", "room", "Renamed", "later") + .unwrap()); + assert!(book + .accept(req.clone(), "other", "room", "Other", "later") + .is_err()); + let mut changed = req; + changed.account_id = "another".into(); + assert!(book + .accept(changed, "caller", "room", "Caller", "later") + .is_err()); + } + #[test] + fn stale_target_does_not_stop_a_new_launch() { + let req = request(Action::Stop); + assert!(req.check_target(Some("old")).is_ok()); + assert!(req.check_target(Some("new")).is_err()); + assert!(req.check_target(None).is_err()); + assert!(request(Action::Start) + .check_target(Some("starting")) + .is_err()); + } + struct Fake { + topic: &'static str, + pty: &'static str, + resume: Result, + calls: Vec, + fail_start: bool, + } + impl Driver for Fake { + fn current(&mut self, req: &Request) -> Result, String> { + if req.topic_id != self.topic || req.account_id != "target" { + return Err("Wrong topic/account".into()); + } + Ok(Some(Current { + pty_id: self.pty.into(), + native_id: Some("native".into()), + resumed_from: None, + })) + } + fn resume(&mut self, _: &Request) -> Result { + self.resume.clone() + } + fn stop(&mut self, req: &Request) -> Result<(), String> { + assert_eq!(req.expected_pty_id.as_deref(), Some(self.pty)); + self.calls.push("stop".into()); + Ok(()) + } + fn start(&mut self, _: &Request, mode: Mode, id: Option<&str>) -> Result { + assert_eq!(mode, Mode::Resume); + assert_eq!(id, Some("native")); + self.calls.push("resume native".into()); + if self.fail_start { + Err("Spawn failed".into()) + } else { + Ok("new".into()) + } + } + } + fn fake() -> Fake { + Fake { + topic: "room", + pty: "old", + resume: Ok("native".into()), + calls: vec![], + fail_start: false, + } + } + #[test] + fn worker_survives_caller_disconnect_and_preserves_resume() { + let (caller, received) = std::sync::mpsc::channel(); + caller.send(request(Action::Restart)).unwrap(); + drop(caller); + let worker = std::thread::spawn(move || { + let req = received.recv().unwrap(); + let mut driver = fake(); + let result = execute(&req, &mut driver); + (result, driver.calls) + }); + let (result, calls) = worker.join().unwrap(); + assert_eq!(result.unwrap(), Some("new".into())); + assert_eq!(calls, ["stop", "resume native"]); + } + #[test] + fn wrong_target_missing_resume_and_launch_failure_are_observable() { + let mut req = request(Action::Restart); + let mut driver = fake(); + req.topic_id = "other".into(); + assert!(execute(&req, &mut driver).is_err()); + assert!(driver.calls.is_empty()); + req.topic_id = "room".into(); + driver.pty = "new"; + assert!(execute(&req, &mut driver).is_err()); + assert!(driver.calls.is_empty()); + driver = fake(); + driver.resume = Err("Transcript missing".into()); + assert!(execute(&req, &mut driver).is_err()); + assert!(driver.calls.is_empty()); + driver = fake(); + driver.resume = Ok("older-native".into()); + assert!(execute(&req, &mut driver).is_err()); + assert!(driver.calls.is_empty()); + driver = fake(); + driver.fail_start = true; + assert_eq!(execute(&req, &mut driver).unwrap_err(), "Spawn failed"); + assert_eq!(driver.calls, ["stop", "resume native"]); + } + #[test] + fn modes_are_explicit_and_restart_cannot_request_fresh() { + let mut req = request(Action::Start); + req.mode = None; + assert!(req.validate().is_err()); + req = request(Action::Restart); + req.mode = Some(Mode::Fresh); + assert!(req.validate().is_err()); + } + #[test] + fn self_restart_record_outlives_caller_and_survives_restart_without_replay() { + let mut book = Book::default(); + { + let caller = String::from("target"); + book.accept(request(Action::Restart), &caller, "room", "Self", "now") + .unwrap(); + } + let mut restored: Book = + serde_json::from_slice(&serde_json::to_vec(&book).unwrap()).unwrap(); + restored.interrupt("restart"); + assert_eq!(restored.operations["op"].status, "interrupted"); + assert!(!restored + .accept(request(Action::Restart), "target", "room", "Self", "later") + .unwrap()); + } +} diff --git a/docs/0-requirements.md b/docs/0-requirements.md index 7daae97..4284a70 100644 --- a/docs/0-requirements.md +++ b/docs/0-requirements.md @@ -142,3 +142,12 @@ Claude の親が利用上限で止まり、拒否が解除予定を名指して ## 投稿結果の時刻と利用量（#385） `say_to_room` は自己呼称に加え、タイムゾーンつき現在時刻と本人の5時間・週間枠（使用率とリセット時刻）、コンテキスト使用率、各データの受信時刻を返す。本文や部屋のログへ流さない。未取得は不明とし、最後の報告を現在値と断定しない。投稿成功・拒否・失敗と last_seen の契約は保つ。利用量の追加読み出しは最大1.5秒で打ち切り、不調でも投稿結果を変更しない。 + + +### ルーム参加 AI によるセッション操作（#415） + +稼働中の認証済み参加アカウントが、設定済みの起動可能アカウントを topic/account/current PTY ID で識別して開始・終了・再起動できる。自分以外の席も対象とし、人間 UI の操作を残す。任意コマンド API にせず、admin と MCP サーバーの起動は拒否する。起動の環境だけに渡す capability を接続で照合し、hello の account 自己申告だけを依頼者認証に使わない。依頼者・厳密な対象・結果／失敗理由を部屋に残す。 + +アプリが受付と結果を永続記録し、自己再起動で呼び出し元が消えても実行を続ける。同じ操作 ID の同一依頼は再実行せず、現行 PTY と異なる終了／再起動を拒否し、UI と起動排他を共有する。通信断は未確認、アプリ終了時の未完了は interrupted とし、次回起動で自動再実行しない。再起動は既存会話の厳格な再開に限定し、再開不能を新規へ置換しない。新規開始は fresh と明示する。 + +状態照会はプロセス起動・現在の稼働・新世代のルーム接続を分ける。Li+ の版／hook 読込確認は本人が起動後に行う。画面の端末と終了イベントを同期し、失敗を観測可能にする。仕様と使用例は [部屋のセッション操作](1-room.md#ai-からのセッション操作415) に定める。 diff --git a/docs/1-room.md b/docs/1-room.md index 39a501d..c220614 100644 --- a/docs/1-room.md +++ b/docs/1-room.md @@ -134,7 +134,7 @@ Pullcept (Tauri app) = 部屋の壁 `history` は参加者からの**読み出しの要求**であり、`history_result` はその答えである。投稿した接続にのみ返す点も、相関を id（`request_id`）で取る点も `post_result` と同じ形である。**これは pull であって push ではない。** 部屋が自分から過去を配ることは無く、席を取る前の発言が届かないことも変わらない——変わるのは、参加者が自分から取りに行けることだけである（後述の「トピック」）。 -未知の `type` は拒否せず無視する。部屋側がフレーム種別を増やしても、旧サイドカーが壊れないため。`protocol` はフレームの形が変わり、サイドカー側が気づく必要があるときに上げる。`say` / `reply` から `post` への統合で `protocol` を 2 へ、`hello` への `hue` の追加で 3 へ、`post` の `last_seen` と `post_result` の追加で 4 へ、`hello` への `account_id` の追加で 5 へ、`history` / `history_result` の追加で 6 へ、`hello` への `room` の追加で 7 へ、部屋から `post` を送らなくなったことで 8 へ（#195）、`to` を一つの名前から名前の並びに変えたことで 9 へ、`usage` / `usage_result` の追加で 10 へ上げた（#364）。サイドカーはアプリが `.mcp.json` へ書く同梱パスから起動するため、部屋とサイドカーの版が食い違う形が構造上無く、旧フレームとの互換は残していない。 +未知の `type` は拒否せず無視する。部屋側がフレーム種別を増やしても、旧サイドカーが壊れないため。`protocol` はフレームの形が変わり、サイドカー側が気づく必要があるときに上げる。`say` / `reply` から `post` への統合で `protocol` を 2 へ、`hello` への `hue` の追加で 3 へ、`post` の `last_seen` と `post_result` の追加で 4 へ、`hello` への `account_id` の追加で 5 へ、`history` / `history_result` の追加で 6 へ、`hello` への `room` の追加で 7 へ、部屋から `post` を送らなくなったことで 8 へ（#195）、`to` を一つの名前から名前の並びに変えたことで 9 へ、`usage` / `usage_result` の追加で 10 へ（#364）、セッション操作と状態照会の追加で 11 へ上げた（#415）。サイドカーはアプリが `.mcp.json` へ書く同梱パスから起動するため、部屋とサイドカーの版が食い違う形が構造上無く、旧フレームとの互換は残していない。 `to` は宛先となる参加者の表示名の並びを入れる（#204）。一人でも並びで送る。サイドカーの `post`、画面へ渡す `room-message`、端末へ打つ札、`post_result` の `missed` と `history_result` の `posts` とで同じ語彙である。任意であり、無いときは `null` や空の並びではなくキーごと省く。部屋は受け取った並びを、各名前の前後の空白を落とし、空の名前と二度目の名前を除いて持つ（`room_floor::addressees`）。除いて何も残らなければ部屋全体宛である。 @@ -539,3 +539,30 @@ liplus-desktop の `stream_parser.rs` および `spawn_stream_pty` / `spawn_stre `say_to_room` の返り値には自己情報に続いて現在時刻（現地時刻と UTC 差）、本人の5時間・週間枠の使用率とリセット時刻、コンテキスト使用率を添える。`my_usage` と同じ読み出し・整形を使い、各値の受信時刻と経過時間を示して最後の報告と明記する。窓・リセット・context の未取得は `not reported`、読み出し不調は `unknown` とする。context は有効な Claude status 報告か Codex の token_count の info から記録し、画面への再送では受信時刻を進めない。context は新しい起動時に捨て、前のセッションの値を引き継がない。枠の値は従来どおり受信時刻つきで保持する。 投稿結果の判定後に利用量を読み、追加の待ちは最大1.5秒とする。成功・拒否・失敗・未確認の結果と last_seen、保持する下書きは変えない。利用量の取得失敗はその追記だけに表し、投稿の isError を変えない。読み出しは本人の接続だけを対象とし、本文・床・ログには追加しない。現在時刻は各呼び出しの返却時に計算し直す。 + + +## AI からのセッション操作（#415） + +認証済みの稼働中アカウントの参加者は、自分と他の席の開始・終了・再起動を依頼できる。呼び出しごとの人間の確認は挟まず、人間の画面の開始・終了は引き続き使える。部屋トークンによる接続認証に加え、アプリが各起動へ環境変数で渡した操作 capability を WebSocket の handshake で照合する。Claude／generic CLI も対象で、hello の account/topic 自己申告だけでは操作権限や依頼者帰属を認めない。capability の値は MCP 登録、argv、一覧、ログに載せず、CLI の子プロセスへの継承と handshake にだけ用いる。外部クライアントの接続だけ、アカウント無し、終了済みの起動の接続からは受け付けない。admin（人間）とローカル MCP サーバーは起動対象にできない。 + +1. `list_account_sessions` で設定済み `account_id`、開いている `topic_id`、現在の `pty_id` を読む。一覧はコマンドや環境変数を返さない。 +2. `control_account_session` に自分で決めた一意の `operation_id` と対象を渡す。開始は `action: "start", mode: "fresh"` または `mode: "resume"` を明示する。終了は `action: "stop"`、再起動は `action: "restart"` と現在の `expected_pty_id` を渡し、mode は渡さない。 +3. `account_session_status` に同じ `operation_id` を渡して確認する。新しい席や別の参加者も照会できる。 + +例（ID は一覧の実値に置き換える）: + +```json +{"operation_id":"restart-luna-20261015-01","action":"restart","account_id":"account-id","topic_id":"topic-id","expected_pty_id":"current-pty-id"} +``` + +再起動は現在の会話 ID とトピックの再開記録・transcript を停止前に照合し、既存の再開処理を使う。未取得 ID、失われた transcript、再開コマンド無しは失敗とし、対象を止めない。停止後に再開条件が変わったり起動が失敗した場合は失敗理由を残す。新規会話への自動置換はしない。画面からの従来の開始にある再開不能時の動作は変えない。 + +アプリは受付を `session-operations.json`（app data）へ保存してから、接続タスクと独立した worker で実行する。同一 ID・同一依頼・同一依頼者の再試行は同じ記録を返し、二度実行しない。同じ ID を他の内容／依頼者に使うと拒否する。別 ID の競合は、UI 起動と共有する lifecycle lock と現在 PTY の照合で処理する。自己再起動の MCP の応答は自分の終了に間に合わないことがある。通信断や15秒の待ち切れは未確認であり、実行中止を意味しない。**新しい ID で送り直さず、同じ ID で照会／再試行する。** + +状態は `accepted` → `running` → `process_started` / `stopped` / `failed`。開始成功はプロセスを作ったという意味で、`process_running` と `room_connected` を別に確認する。接続は同じアカウント名ではなく起動世代と接続で照合する。`stopped` は Pullcept の PTY 所有の解放と終了要求を示し、OS 上のプロセス消滅の検証ではない。再起動成功だけで Li+ 版や hook 読込を反映済みとは扱わず、新しい席本人が確認する。 + +依頼者、対象 topic/account/expected PTY、操作、結果／理由は依頼者の部屋にアプリ通知として残し、画面にも失敗を出す。MCP から作った端末も既存の seat 読み直しで採用する。採用前の出力や旧端末の scrollback は再構築しない（画面再読込時と同じ制約）。 + +アプリ終了時と次回起動時に未完了の `accepted` / `running` を `interrupted` にし、自動再実行しない。強制終了時は最後に保存できた状態から判断する。完了保存が失敗した場合は画面のログエラーを確認し、実際の席を一覧で照会する。記録は保持し続けるため、app data の削除は重複排除の履歴も消す。自動の異常検出・相互停止規約・Li+ 更新・版固定はこの機能に含めない。 + +wire protocol 11 は `session_list`、`session_status`（`operation_id`）、`session_control`（`operation`）を送り、`session_result`（`request_id` と `result` または `error`）を返す。操作IDは永続の重複排除、request ID は接続上の応答照合に用いる。 diff --git a/docs/3-accounts.md b/docs/3-accounts.md index 572a521..1ced79c 100644 --- a/docs/3-accounts.md +++ b/docs/3-accounts.md @@ -4,7 +4,7 @@ 種別で `Codex CLI` を選び、CLI コマンド（`codex` または実行ファイルの絶対パス）と作業ディレクトリを設定する。対応基準は **Codex CLI 0.160.0 以上**であり、古い CLI では起動前に診断を返す。アプリはインストール・認証・利用者のグローバル設定を変更しない。宣言済み `cli` は command が codex でも汎用のままであり、移行対象は旧 `ai` または kind の無いアカウントだけである。 -**Codex の部屋の MCP は起動の行だけで渡し、ファイルには書かない（#297）。** 起動の `-c mcp_servers.<自席>={…}` が定義全体（コマンド・引数・`env_vars`・`enabled=true`・二道具 `say_to_room` と `read_room_history` の承認）を運ぶ。席の名前は Claude の席と同じアカウント／トピックごとの名前（`server_name_for`）である。席の属性（部屋 URL・名前・アカウント・部屋 id・色・未読の印）と部屋トークンは起動した CLI の環境にあり、定義は名前だけを `env_vars` で指す。部屋トークンは設定・argv・プレビューに値を残さない。承認を起動時に指定するのは二道具に限る。行に載るのは自席の定義だけで、他の席の名前は載らない——Codex の他の席の部屋サーバーはそれぞれの起動の行にしか無く、この席からは見えないため、止める対象が無い。起動の前に有効 cwd に `.codex` フォルダーだけを用意する。Codex 0.160.0 の native discovery は `config.toml` の無い `.codex` フォルダーでも project layer を返し（[loader](https://github.com/openai/codex/blob/rust-v0.160.0/codex-rs/config/src/loader/mod.rs) の `discover_project_layers`）、専用 hook の置き場はその layer から決まる（下記）。 +**Codex の部屋の MCP は起動の行だけで渡し、ファイルには書かない（#297）。** 起動の `-c mcp_servers.<自席>={…}` が定義全体（コマンド・引数・`env_vars`・`enabled=true`・部屋の六道具の承認）を運ぶ。席の名前は Claude の席と同じアカウント／トピックごとの名前（`server_name_for`）である。席の属性（部屋 URL・名前・アカウント・部屋 id・色・未読の印）と部屋トークンは起動した CLI の環境にあり、定義は名前だけを `env_vars` で指す。部屋トークンと起動別のセッション操作 capability（#415）は設定・argv・プレビューに値を残さない。発言・履歴・利用量と、[セッション操作の三道具](1-room.md#ai-からのセッション操作415)を起動時に承認する。行に載るのは自席の定義だけで、他の席の名前は載らない——Codex の他の席の部屋サーバーはそれぞれの起動の行にしか無く、この席からは見えないため、無効化する他席の MCP 登録が無い。起動の前に有効 cwd に `.codex` フォルダーだけを用意する。Codex 0.160.0 の native discovery は `config.toml` の無い `.codex` フォルダーでも project layer を返し（[loader](https://github.com/openai/codex/blob/rust-v0.160.0/codex-rs/config/src/loader/mod.rs) の `discover_project_layers`）、専用 hook の置き場はその layer から決まる（下記）。 **以前の版が `.codex/config.toml` に書いた部屋の登録は、その席が起動するときに消す（#297）。** 消すのは名前が `pullcept-room` で始まり、`env` に `PULLCEPT_ACCOUNT_ID` を持つ欄だけで、部屋 URL を問わない。他のキー・欄・コメントは残し、消す欄が無ければファイルを書き直さない。欄を消して `[mcp_servers]` が空になればその表も外す。ファイルそのものは消さない。対象は起動する有効 cwd のファイルだけであり、別のフォルダーに残った欄は、そのフォルダーで席が起動したときに消える。残っていても無効（`enabled = false`）で、サーバーは起動しない。消してよいのは、#272 以来どの版も同じ定義全体を起動の行で渡しており、ファイルの欄は既定で無効のまま一度もサーバーを起動していないためである。 @@ -411,13 +411,13 @@ Windows ではセッションの起動が `cmd.exe /C` を通る（`src-tauri/sr **席の台帳が起動そのものの記録を持つ。** `pty_id` に加えて、開始時刻・起動コマンド・作業ディレクトリを持つ。アカウント id から `pty_id` への対応表は初めから台帳に在り、`seated_accounts` が鍵だけを返していた——在るものを渡していなかった。残り三つを台帳へ置くのは、アカウントが稼働中でも編集できるためである。アカウントから読むと、いま走っているものではなく、起動の後に書き換えられた値を出すことになる。 -**画面は席を見て端末を作り直す。** 席を持つ（トピック, アカウント）の組に端末が無ければ、その `pty_id` で `pty-data-{id}` と `pty-exit-{id}` を購読し直す。出力も入力も終了も戻る。席を強制的に空ける経路は作らない——プロセスが在るのに席だけ空けば二重起動が通る（上記「一つのアカウントは一つの部屋に一席まで」）。台帳はアプリの記憶であるから、この繋ぎ直しが起きるのは webview の再読み込みの後だけであって、アプリの再起動の後ではない。 +**画面は席を見て端末を作り直す。** 席を持つ（トピック, アカウント）の組に端末が無ければ、その `pty_id` で `pty-data-{id}` と `pty-exit-{id}` を購読し直す。出力も入力も終了も戻る。席を強制的に空ける経路は作らない——プロセスが在るのに席だけ空けば二重起動が通る（上記「一つのアカウントは一つの部屋に一席まで」）。台帳はアプリの記憶であるから、アプリの再起動の後には稼働中の席が残らない。webview の再読み込みに加え、MCP の開始・再起動後も同じ採用経路を使う（#415）。同じ組に古い PTY の端末が残っていれば新しい PTY の端末へ置き換える。 起動が飛んでいる最中に再読み込みが落ちた場合、その席には繋ぎ直す先がまだ無い（席は確保済みで `pty_id` は未定）。行の `▶️` を押すと台帳を読み直し、その時点で起動が済んでいれば端末が戻る。拒否は読み直しの後に出す。 -**線はスクロールバックで引く。** 過去の出力は消えた xterm のインスタンスに在り、取り戻す経路が無い。再読み込みから繋ぎ直しまでの出力も同様に落ちる。繋ぎ直した端末は、そこで出力が切れていることを 1 行目に書いてから開く。空の端末が「起動中」と並ぶと、何も出力していないセッションに読める——落ちる直前の出力を読んで判断する形（「[診断面](2-screen.md#診断面)」、#57）は、その 1 行が無いと成り立たない。 +**線はスクロールバックで引く。** 過去の出力は消えた xterm のインスタンスに在り、取り戻す経路が無い。再読み込みから繋ぎ直しまでの出力も同様に落ちる。採用した端末は、接続より前の出力を復元できないことを 1 行目と状態行で知らせる。webview の再読み込み後は従来どおり診断ペインを開き、MCP の開始・再起動ではペインの開閉を保つ（#415）。空の端末が「起動中」と並ぶと、何も出力していないセッションに読める——落ちる直前の出力を読んで判断する形（「[診断面](2-screen.md#診断面)」、#57）は、その 1 行が無いと成り立たない。 -**ただし、その 1 行は実機では読めなかった**（2026-08-24、AI が操作）。繋ぎ直した直後に Claude Code が全画面を描き直すため、書いた行はその再描画に上書きされる。告知そのものは状態行にも出ており、そちらは残った——「〈名前〉 の端末に繋ぎ直しました。再読み込みより前の出力は残っていません。」である。情報は伝わっているが、伝えているのは端末の 1 行目ではない。以前ここに「告知が端末の 1 行目で読める」と読める書き方をしていたのは実態と違う。行そのものは書いたまま残す——全画面を描き直さない CLI ではその位置で読めるためであり、消せばそちらで失われる。 +**ただし、その 1 行は実機では読めなかった**（2026-08-24、AI が操作）。繋ぎ直した直後に Claude Code が全画面を描き直すため、書いた行はその再描画に上書きされる。告知そのものは状態行にも出ており、そちらは残った——「〈名前〉 の端末に繋ぎ直しました。再読み込みより前の出力は残っていません。」である。情報は伝わっているが、伝えているのは端末の 1 行目ではない。以前ここに「告知が端末の 1 行目で読める」と読める書き方をしていたのは実態と違う。行そのものは書いたまま残す——全画面を描き直さない CLI ではその位置で読めるためであり、消せばそちらで失われる。 #415 から現在の告知は「稼働中の端末に接続／接続より前の出力は復元されません」とし、再読み込みが無い MCP 操作にも同じ意味で使う。 **繋ぎ直した端末は、その席のトピックの下に置く（#141）。** 画面に出ていないトピックの端末は隠したまま作り、そのトピックを開いたときに出る。 diff --git a/sidecar/src/index.ts b/sidecar/src/index.ts index ba56d1e..6716ce6 100644 --- a/sidecar/src/index.ts +++ b/sidecar/src/index.ts @@ -173,7 +173,7 @@ if (SEAT_REFUSAL === null && ADMISSION_REQUIRED && */ const UNSEEN_HISTORY = process.env.PULLCEPT_UNSEEN_HISTORY === "1"; -const PROTOCOL_VERSION = 10; +const PROTOCOL_VERSION = 11; /** * What the first line of a post typed into this session's terminal opens with @@ -562,6 +562,30 @@ const TOOLS = [ required: [] as string[], }, }, + { + name: "list_account_sessions", + description: "List configured account IDs, launchable kinds, open topic IDs, current PTY IDs and room connection state. No commands or environment secrets. Authenticated room participants may control their own and other sessions.", + inputSchema: { type: "object" as const, properties: {}, required: [] as string[] }, + }, + { + name: "account_session_status", + description: "Read a persisted operation by operation_id. process_started means launch only; check process_running and room_connected separately. App exit marks unfinished operations interrupted; never replays them. Restart does not prove Li+ version or hooks loaded: the new seat must verify those itself.", + inputSchema: { type: "object" as const, properties: { operation_id: { type: "string" } }, required: ["operation_id"] }, + }, + { + name: "control_account_session", + description: "Ask Pullcept to start, stop or restart a configured account in an open topic, including yourself or another seat. Supply your own unique operation_id BEFORE calling; identical retries with that ID are not repeated. start requires explicit fresh/resume mode; stop/restart require the current expected_pty_id from list_account_sessions and no mode. Restart strictly resumes; unavailable conversation fails, never substitutes fresh. App owns execution after acceptance even if your sidecar dies. Timeout/disconnect means unconfirmed: query or retry the SAME ID. No arbitrary commands. Requester, exact target, result/error are recorded in the room.", + inputSchema: { + type: "object" as const, + properties: { + operation_id: { type: "string", minLength: 1, maxLength: 200 }, + action: { type: "string", enum: ["start", "stop", "restart"] }, + account_id: { type: "string" }, topic_id: { type: "string" }, + expected_pty_id: { type: "string" }, mode: { type: "string", enum: ["fresh", "resume"] }, + }, + required: ["operation_id", "action", "account_id", "topic_id"], additionalProperties: false, + }, + }, ]; /** @@ -599,6 +623,15 @@ mcp.setRequestHandler(CallToolRequestSchema, async (request) => { if (name === "read_room_history") return await readHistory(args); if (name === "my_usage") return await readUsage(); + if (name === "list_account_sessions") return await sessionRequest("session_list", {}); + if (name === "account_session_status") { + if (typeof args?.operation_id !== "string" || !args.operation_id.trim()) return sessionError("operation_id required"); + return await sessionRequest("session_status", { operation_id: args.operation_id }); + } + if (name === "control_account_session") { + if (!args || typeof args.operation_id !== "string" || !args.operation_id.trim()) return sessionError("Supply operation_id before calling; keep it for retries and status"); + return await sessionRequest("session_control", { operation: args }); + } if (name !== "say_to_room") { return { @@ -1005,6 +1038,23 @@ const awaitingResult = new Map void> */ const awaitingHistory = new Map void>(); +interface SessionResultFrame { type: "session_result"; request_id?: string; result?: unknown; error?: string; } +const awaitingSessions = new Map void>(); +function sessionError(text: string) { return { content: [{ type: "text" as const, text }], isError: true }; } +async function sessionRequest(type: string, fields: Record) { + const requestId = randomUUID(); + const answer = new Promise((resolve) => { + const timer = setTimeout(() => { awaitingSessions.delete(requestId); resolve(null); }, POST_RESULT_TIMEOUT); + timer.unref?.(); + awaitingSessions.set(requestId, (result) => { clearTimeout(timer); awaitingSessions.delete(requestId); resolve(result); }); + }); + if (!sendToRoom({ type, request_id: requestId, ...fields })) awaitingSessions.get(requestId)?.(null); + const result = await answer; + if (result === null) return sessionError("Session request unconfirmed (disconnected or timed out). Query operation_id or retry with the SAME operation_id; do not create a new ID."); + if (result.error) return sessionError(result.error); + return { content: [{ type: "text" as const, text: JSON.stringify(result.result) }] }; +} + /** Reads of this seat's usage waiting for the room's answer, by request id. */ const awaitingUsage = new Map void>(); @@ -1096,6 +1146,7 @@ function abandonPendingPosts(): void { for (const settle of [...awaitingResult.values()]) settle(null); for (const settle of [...awaitingHistory.values()]) settle(null); for (const settle of [...awaitingUsage.values()]) settle(null); + for (const settle of [...awaitingSessions.values()]) settle(null); } function sendToRoom(frame: Record): boolean { @@ -1126,6 +1177,13 @@ function connectRoom(): void { headers["x-pullcept-room"] = ROOM_ID; } + const capability = process.env.PULLCEPT_CONTROL_CAPABILITY?.trim(); + if (capability && ACCOUNT_ID && ROOM_ID) { + headers["x-pullcept-control"] = capability; + headers["x-pullcept-control-account"] = ACCOUNT_ID; + headers["x-pullcept-control-room"] = ROOM_ID; + } + const socket = new WebSocket(ROOM_URL, { headers }); ws = socket; @@ -1173,6 +1231,10 @@ function connectRoom(): void { // in order not to do. else if (frame.type === "history_result") settleHistoryResult(frame as HistoryResultFrame); else if (frame.type === "usage_result") settleUsageResult(frame as UsageResultFrame); + else if (frame.type === "session_result") { + const result = frame as SessionResultFrame; + if (typeof result.request_id === "string") awaitingSessions.get(result.request_id)?.(result); + } // Unknown frame kinds are ignored on purpose; see the frame comment above. // A `post` frame from a room older than protocol 8 is one of them: the // session is typed its posts by the app, and there is nowhere here to put diff --git a/sidecar/test/admission.test.mjs b/sidecar/test/admission.test.mjs index 295844a..cdc05c4 100644 --- a/sidecar/test/admission.test.mjs +++ b/sidecar/test/admission.test.mjs @@ -124,7 +124,7 @@ test('real MCP parent admitted before initialize, child empty, same sidecar reco launcher.ledger.values().next().value.pending = 2; const parent = await sidecar(t, launcher.port, launcher.launch); assert.match(parent.initialized.result.instructions, /Pullcept room/); - assert.equal((await parent.rpc('tools/list')).result.tools.length, 3); + assert.equal((await parent.rpc('tools/list')).result.tools.length, 6); await until(() => launcher.connections.length === 1); const instance = launcher.connections[0].headers['x-pullcept-instance']; assert.equal(instance, launcher.requests[0].instance_id); @@ -138,7 +138,7 @@ test('real MCP parent admitted before initialize, child empty, same sidecar reco launcher.connections[0].socket.close(); await until(() => launcher.connections.length === 2); assert.equal(launcher.connections[1].headers['x-pullcept-instance'], instance); - assert.equal((await parent.rpc('tools/list')).result.tools.length, 3); + assert.equal((await parent.rpc('tools/list')).result.tools.length, 6); assert.ok(!child.stderr().includes('synthetic-only')); }); @@ -153,7 +153,7 @@ test('stale/dead launches rejected and distinct topics/accounts admitted', async for (const [account, topic] of [['synthetic-account', 'topic-2'], ['account-2', 'synthetic-topic']]) { const launch = randomUUID(); launcher.ledger.set(`${topic}/${account}`, { launch, live: true, owner: null }); const parent = await sidecar(t, launcher.port, launch, account, topic); - assert.equal((await parent.rpc('tools/list')).result.tools.length, 3); + assert.equal((await parent.rpc('tools/list')).result.tools.length, 6); } await until(() => launcher.connections.length === 2); assert.equal(launcher.connections.length, 2); @@ -165,7 +165,7 @@ test('legacy Claude and no-account sidecars keep instructions/tools without admi for (const account of ['synthetic-account', '']) { const legacy = await sidecar(t, launcher.port, '', account, 'synthetic-topic', false); assert.match(legacy.initialized.result.instructions, /Pullcept room/); - assert.equal((await legacy.rpc('tools/list')).result.tools.length, 3); + assert.equal((await legacy.rpc('tools/list')).result.tools.length, 6); } await until(() => launcher.connections.length === 2); assert.equal(launcher.requests.length, 0); diff --git a/sidecar/test/round-trip.test.mjs b/sidecar/test/round-trip.test.mjs index e391903..4d19a84 100644 --- a/sidecar/test/round-trip.test.mjs +++ b/sidecar/test/round-trip.test.mjs @@ -299,6 +299,8 @@ test("say_to_room reaches the room, and the room pushes nothing back", async (t) }, ]; const historyFrames = []; + const sessionFrames = []; + let sessionAnswer = "normal"; // What the room answers the next reads of this seat's usage with (#364), // in order: figures, then none received yet, then a room that cannot say @@ -316,12 +318,25 @@ test("say_to_room reaches the room, and the room pushes nothing back", async (t) let explicitUsage = false; let feedbackUsage = { five_hour: { used_percentage: 78, resets_at: 1791207909 }, seven_day: { used_percentage: 12 }, received_at: RECEIVED, context: 42, context_received_at: RECEIVED }; - wss.on("connection", (socket) => { + wss.on("connection", (socket, req) => { + assert.equal(req.headers["x-pullcept-control"], "synthetic-control-only"); + assert.equal(req.headers["x-pullcept-control-account"], TEST_ACCOUNT); + assert.equal(req.headers["x-pullcept-control-room"], "test-room"); roomSocket = socket; connected.resolve(socket); socket.on("message", (raw) => { const frame = JSON.parse(raw.toString()); if (frame.type === "hello") helloSeen.resolve(frame); + if (["session_list", "session_status", "session_control"].includes(frame.type)) { + sessionFrames.push(frame); + socket.send(JSON.stringify({ type: "session_result", request_id: "unrelated", result: { wrong: true } })); + if (sessionAnswer === "disconnect") { socket.close(); return; } + socket.send(JSON.stringify({ type: "session_result", request_id: frame.request_id, + ...(sessionAnswer === "denied" ? { error: "Only a connected account session may control sessions" } : + { result: frame.type === "session_list" ? { accounts: [{ account_id: TEST_ACCOUNT }], topics: [{ topic_id: "room" }] } : + { operation_id: frame.operation?.operation_id ?? frame.operation_id, status: "accepted", room_connected: false } }) })); + return; + } if (frame.type === "usage") { usageFrames.push(frame); if (!explicitUsage && feedbackUsage === null) return; @@ -431,8 +446,12 @@ test("say_to_room reaches the room, and the room pushes nothing back", async (t) ...process.env, PULLCEPT_LAUNCH_ID: "", PULLCEPT_ROOM_ADMISSION: "0", + PULLCEPT_CONTROL_CAPABILITY: "", PULLCEPT_ROOM_URL: `ws://127.0.0.1:${port}`, PULLCEPT_AGENT_NAME: "test-agent", + PULLCEPT_CONTROL_CAPABILITY: "synthetic-control-only", + PULLCEPT_ACCOUNT_NAME: "test-agent", + PULLCEPT_UNSEEN_HISTORY: "0", PULLCEPT_AGENT_HUE: "145", PULLCEPT_ACCOUNT_ID: TEST_ACCOUNT, // Launched as the account its entry names, which is what every launch @@ -594,8 +613,8 @@ test("say_to_room reaches the room, and the room pushes nothing back", async (t) // `my_usage` reads too, and only this seat's own figures (#364). assert.deepEqual( toolNames, - ["say_to_room", "read_room_history", "my_usage"], - "one posting tool and two reading tools, and nothing else", + ["say_to_room", "read_room_history", "my_usage", "list_account_sessions", "account_session_status", "control_account_session"], + "one posting tool plus room reads and account lifecycle controls", ); assert.equal( toolNames.filter((name) => name === "say_to_room").length, @@ -624,6 +643,25 @@ test("say_to_room reaches the room, and the room pushes nothing back", async (t) // The manners that moved off the instructions (#273), each asserted whole // where it now lives. Nothing here may be lost to the move: these are the // lines the 2048 cut had taken from every session. + assert.ok(!JSON.stringify(tools).includes("synthetic-control-only"), "launch capability must not appear in tools/list"); + const listed = await request("tools/call", { name: "list_account_sessions", arguments: {} }); + assert.match(listed.result.content[0].text, new RegExp(TEST_ACCOUNT)); + const operation = { operation_id: "self-restart-415", action: "restart", account_id: TEST_ACCOUNT, topic_id: "room", expected_pty_id: "old" }; + const accepted = await request("tools/call", { name: "control_account_session", arguments: operation }); + assert.match(accepted.result.content[0].text, /accepted/); + assert.deepEqual(sessionFrames.at(-1).operation, operation); + const retry = await request("tools/call", { name: "control_account_session", arguments: operation }); + assert.match(retry.result.content[0].text, /self-restart-415/); + const checked = await request("tools/call", { name: "account_session_status", arguments: { operation_id: operation.operation_id } }); + assert.match(checked.result.content[0].text, /self-restart-415/); + sessionAnswer = "denied"; + const denied = await request("tools/call", { name: "list_account_sessions", arguments: {} }); + assert.equal(denied.result.isError, true); assert.match(denied.result.content[0].text, /connected account/); + sessionAnswer = "normal"; + const beforeMissingId = sessionFrames.length; + const missingId = await request("tools/call", { name: "control_account_session", arguments: { action: "restart" } }); + assert.equal(missingId.result.isError, true); assert.equal(sessionFrames.length, beforeMissingId); + const say = tools.result.tools.find((tool) => tool.name === "say_to_room"); assertContains(say.description, REPLY, "say_to_room must say it is the only way to be heard"); assertContains( @@ -684,7 +722,7 @@ test("say_to_room reaches the room, and the room pushes nothing back", async (t) // The room it was started into. One socket serves every topic open in the // app, so this is what puts the connection in one of them (#141). assert.equal(hello.room, "test-room"); - assert.equal(hello.protocol, 10); + assert.equal(hello.protocol, 11); // A `post` frame, as a room older than protocol 8 would send. The session // is typed its posts by the app now (#195): this process pushes nothing into @@ -1045,7 +1083,11 @@ test("say_to_room reaches the room, and the room pushes nothing back", async (t) "still waiting", "say_to_room must not resolve before the room answers; a call that returns on send is not a boundary", ); - roomSocket.close(); + sessionAnswer = "disconnect"; + const operationUnconfirmed = await request("tools/call", { name: "control_account_session", arguments: operation }); + assert.equal(operationUnconfirmed.result.isError, true); + assert.match(operationUnconfirmed.result.content[0].text, /unconfirmed/); + assert.match(operationUnconfirmed.result.content[0].text, /SAME operation_id/); const unconfirmed = await unanswered; assert.ok( unconfirmed.result.isError, @@ -1077,6 +1119,8 @@ test("say_to_room reaches the room, and the room pushes nothing back", async (t) "Not delivered: the room socket is not connected", "a send with no socket must say so rather than wait out the answer it will never get", ); + + }); test("a session launched without a hue or an account says so by omission", async (t) => { @@ -1099,6 +1143,7 @@ test("a session launched without a hue or an account says so by omission", async socket.on("message", (raw) => { const frame = JSON.parse(raw.toString()); if (frame.type === "hello") helloSeen.resolve(frame); + }); }); @@ -1111,6 +1156,7 @@ test("a session launched without a hue or an account says so by omission", async ...process.env, PULLCEPT_LAUNCH_ID: "", PULLCEPT_ROOM_ADMISSION: "0", + PULLCEPT_CONTROL_CAPABILITY: "", PULLCEPT_ROOM_URL: `ws://127.0.0.1:${port}`, PULLCEPT_AGENT_NAME: "no-colour", PULLCEPT_AGENT_HUE: "", @@ -1163,6 +1209,7 @@ test("a session seated in a topic that already holds posts is told so", async (t ...process.env, PULLCEPT_LAUNCH_ID: "", PULLCEPT_ROOM_ADMISSION: "0", + PULLCEPT_CONTROL_CAPABILITY: "", // Unset on purpose: no room to connect to, and the sidecar stays // offline and serving rather than exiting. PULLCEPT_ROOM_URL: "", @@ -1272,6 +1319,7 @@ for (const [label, launchedAs, launchedRoom] of [ ...process.env, PULLCEPT_LAUNCH_ID: "", PULLCEPT_ROOM_ADMISSION: "0", + PULLCEPT_CONTROL_CAPABILITY: "", PULLCEPT_ROOM_URL: `ws://127.0.0.1:${port}`, PULLCEPT_AGENT_NAME: "test-agent", PULLCEPT_ACCOUNT_ID: TEST_ACCOUNT, diff --git a/src-tauri/Cargo.lock b/src-tauri/Cargo.lock index ccb20c5..4301697 100644 --- a/src-tauri/Cargo.lock +++ b/src-tauri/Cargo.lock @@ -2183,6 +2183,7 @@ name = "mcp-config" version = "0.1.0" dependencies = [ "chrono", + "serde", "serde_json", "toml_edit 0.20.2", ] diff --git a/src-tauri/src/codex_app_server.rs b/src-tauri/src/codex_app_server.rs index f512f86..13852ce 100644 --- a/src-tauri/src/codex_app_server.rs +++ b/src-tauri/src/codex_app_server.rs @@ -102,6 +102,11 @@ fn stop_tree(child: &mut Child) { } /// End every seat server, for the app being closed. +pub fn stop_one(app: &AppHandle, pty_id: &str) { + let server = app.state::().servers.lock().remove(pty_id); + if let Some(server) = server { server.stop(); } +} + pub fn stop_all(app: &AppHandle) { let drained: Vec = app .state::() diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 0ed799c..8b16919 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -11,6 +11,7 @@ mod pty; mod room; mod room_log; mod session; +mod session_control; mod webhook; #[cfg(windows)] mod window_icon; @@ -31,6 +32,7 @@ pub fn run() { // Which account is in the room, so a second launch of one account is // refused rather than seating one identity twice (session::RoomSeats). .manage(RoomSeats::new()) + .manage(session_control::SessionControl::default()) // The MCP servers the app runs itself, by name (#172). .manage(McpServers::new()) // The Codex seats stopped on their usage limit, and the posts the room @@ -45,6 +47,9 @@ pub fn run() { // What each Claude Code seat is doing, as its hooks tell it (#331). .manage(hook_activity::HookSeats::new()) .setup(|app| { + if let Err(error) = app.state::().initialize(app.handle()) { + eprintln!("[session-control] {error}"); + } #[cfg(windows)] for window in app.webview_windows().values() { let result = window @@ -128,6 +133,10 @@ pub fn run() { .run(|app, event| { // However the app is left, no seat's app-server outlives it (#299). if let tauri::RunEvent::Exit = event { + app.state::().close(app); + let seats = app.state::(); + let _guard = seats.lifecycle.lock(); + app.state::().kill_all(); codex_app_server::stop_all(app); } }); diff --git a/src-tauri/src/pty.rs b/src-tauri/src/pty.rs index 34998f6..a4cf099 100644 --- a/src-tauri/src/pty.rs +++ b/src-tauri/src/pty.rs @@ -161,6 +161,13 @@ impl PtyState { self.procs.lock().contains_key(id) } + pub fn kill_exact(&self, id: &str) -> Result<(), String> { + if let Some(mut pty) = self.procs.lock().remove(id) { + pty.child.kill().map_err(|e| format!("Kill failed: {e}"))?; + } + Ok(()) + } + /// Kill every session that is running. /// /// One acquisition of the lock over the whole map, for the same reason @@ -489,12 +496,10 @@ pub fn resize_pty( } #[tauri::command] -pub fn kill_pty(state: tauri::State, id: String) -> Result<(), String> { - let mut map = state.procs.lock(); - if let Some(mut pty) = map.remove(&id) { - pty.child.kill().map_err(|e| format!("Kill failed: {e}"))?; - } - Ok(()) +pub fn kill_pty(app: AppHandle, state: tauri::State, id: String) -> Result<(), String> { + let seats = app.state::(); + let _guard = seats.lifecycle.lock(); + state.kill_exact(&id) } /// End every running session at once, for the app being closed (#85). @@ -522,6 +527,9 @@ pub fn kill_pty(state: tauri::State, id: String) -> Result<(), String> /// wait for (#299). #[tauri::command] pub fn kill_all_ptys(app: AppHandle, state: tauri::State) { + app.state::().close(&app); + let seats = app.state::(); + let _guard = seats.lifecycle.lock(); state.kill_all(); crate::codex_app_server::stop_all(&app); } diff --git a/src-tauri/src/room.rs b/src-tauri/src/room.rs index 32a1435..b76339a 100644 --- a/src-tauri/src/room.rs +++ b/src-tauri/src/room.rs @@ -182,7 +182,8 @@ use uuid::Uuid; /// 9: `post` carries `to` as a list of names (#204). /// 10: `usage` / `usage_result` — a participant may read its own seat's /// usage-limit figures (#364). Its own only, read off the connection. -pub const PROTOCOL_VERSION: u32 = 10; +/// 11: app-owned session operations and operation status (#415). +pub const PROTOCOL_VERSION: u32 = 11; /// One post of the room, as the frontend sees it. /// @@ -322,6 +323,8 @@ struct IncomingFrame { request_id: Option, limit: Option, before: Option, + operation_id: Option, + operation: Option, } /// One room: a topic, the floor of what has been said in it this run, and who @@ -615,6 +618,10 @@ impl RoomState { .map(|room| room.topic.clone()) } + pub fn open_topics(&self) -> Vec { + self.inner.lock().rooms.values().map(|room| room.topic.clone()).collect() + } + /// The topic the screen has open. pub fn topic(&self) -> TopicRef { let inner = self.inner.lock(); @@ -1852,6 +1859,7 @@ async fn serve_participant( // The token is what makes this room, and not merely this machine. let socket_id = Uuid::new_v4().to_string(); let mut admitted: Option<(mcp_config::codex::admission::Claim, String)> = None; + let mut control_bound: Option = None; let check = |req: &Request, res: Response| -> Result { let ok = req .headers() @@ -1860,6 +1868,19 @@ async fn serve_participant( .map(|v| v == expected) .unwrap_or(false); if ok { + if req.headers().contains_key("x-pullcept-control") { + let field = |key| req.headers().get(key).and_then(|v| v.to_str().ok()); + let binding = field("x-pullcept-control-room").zip(field("x-pullcept-control-account")).zip(field("x-pullcept-control")); + match binding { + Some(((topic, account), capability)) if app.state::().control_binding(topic, account, capability, &app.state::()) => { + control_bound = Some(mcp_config::session_control::ControlBinding { topic: topic.into(), account: account.into(), capability: capability.into() }); + }, + _ => { + let mut deny = ErrorResponse::new(Some("session control capability rejected".into())); + *deny.status_mut() = StatusCode::FORBIDDEN; return Err(deny); + } + } + } if req.headers().contains_key("x-pullcept-launch") { let header = |key| req.headers().get(key).and_then(|v| v.to_str().ok()); let body = serde_json::json!({ @@ -2013,6 +2034,10 @@ async fn serve_participant( }), }; if !valid { break; } + if control_bound.as_ref().is_some_and(|binding| { + !binding.matches_hello(declared_room, declared_account) || + !app.state::().control_binding(&binding.topic, &binding.account, &binding.capability, &app.state::()) + }) { break; } let name = frame.name.unwrap_or_else(|| "session".to_string()); if frame.protocol != Some(PROTOCOL_VERSION) { // Legible mismatch beats a silent half-working room. @@ -2061,6 +2086,10 @@ async fn serve_participant( account.as_deref(), ); if let Ok(config) = config::read_config(&app) { room.refresh_accounts(&app, &config.accounts); } + if let (Some(account), Some(binding)) = (account.as_ref(), control_bound.as_ref()) { + app.state::().connected(&origin, &socket_id, &room_id, account, &binding.capability, + admitted_claim.as_ref().map(|claim| claim.launch_id.as_str())); + } joined_as = Some(name); joined_account = account; let _ = app.emit("room-participants", room.roster(&room_id)); @@ -2203,11 +2232,33 @@ async fn serve_participant( frame: answer.to_string(), }); } + "session_list" | "session_status" | "session_control" => { + let request_id = frame.request_id.unwrap_or_default(); + let control = app.state::(); + let result = (|| -> Result { + control.authorize(&app, &origin, &socket_id)?; + match frame.kind.as_str() { + "session_list" => control.list(&app), + "session_status" => control.status(&app, frame.operation_id.as_deref().ok_or("operation_id required")?), + _ => { + let request = serde_json::from_value(frame.operation.ok_or("operation required")?).map_err(|e| format!("Invalid operation: {e}"))?; + control.accept(&app, request, joined_account.as_deref().ok_or("No requester account")?, + joined_room.get().ok_or("No requester topic")?, joined_as.as_deref().ok_or("No requester name")?) + } + } + })(); + let answer = match result { + Ok(result) => serde_json::json!({ "type": "session_result", "request_id": request_id, "result": result }), + Err(error) => serde_json::json!({ "type": "session_result", "request_id": request_id, "error": error }), + }; + let _ = room.to_participants.send(Fanout { target: origin.clone(), frame: answer.to_string() }); + } _ => {} } } pump.abort(); + app.state::().disconnected(&origin, &socket_id); if joined_as.is_some() && admitted_claim.as_ref().is_none_or(|claim| app.state::().may_unseat_sidecar(claim, &socket_id)) { // By connection. Removing by name took every participant answering to // that name off the roster, so one session ending emptied the other's @@ -2342,6 +2393,7 @@ pub fn room_delete_topic( seats: tauri::State, topic_id: String, ) -> Result, String> { + let _guard = seats.lifecycle.lock(); pty_state.kill_each(&seats.running_in_topic(&topic_id, &pty_state)); room_log::delete_topic(&app, &topic_id)?; diff --git a/src-tauri/src/session.rs b/src-tauri/src/session.rs index 55d7ba3..07250a2 100644 --- a/src-tauri/src/session.rs +++ b/src-tauri/src/session.rs @@ -482,6 +482,7 @@ fn status_command(port: Option, topic_id: &str, account_id: &str) -> Option /// its meaning. #[derive(Clone)] pub struct RoomSeats { + pub lifecycle: Arc>, seats: Arc>>, } @@ -501,7 +502,7 @@ enum Seat { /// being spawned. Held so that two launches racing for one account cannot /// both find the seat empty — the loser is refused before a second process /// exists, rather than after. - Starting { requires_admission: bool, sidecar: Option }, + Starting { generation: String, requires_admission: bool, sidecar: Option }, /// A session is running. The PTY id is what says so, and asking the PTY is /// the only liveness question anyone asks here. Running(RunningSession), @@ -534,6 +535,8 @@ impl StartingSidecar { /// would use now, not what this one used. #[derive(Clone, serde::Serialize)] pub struct RunningSession { + #[serde(skip)] + generation: String, #[serde(skip)] launch_id: Option, #[serde(skip)] @@ -598,7 +601,7 @@ impl RoomSeats { // App-server prepares MCP before it has a PTY. Bind only after its process // exists, and carry this owner into Running rather than reopening admission. fn starting_sidecar(&self, topic: &str, account: &str, launch: &str, child: Arc>) { - if let Some(Seat::Starting { requires_admission: true, sidecar: slot }) = self.seats.lock().get_mut(&seat_key(topic, account)) { + if let Some(Seat::Starting { requires_admission: true, sidecar: slot, .. }) = self.seats.lock().get_mut(&seat_key(topic, account)) { *slot = Some(StartingSidecar { launch_id: launch.into(), child, sidecar: Default::default() }); } } @@ -606,7 +609,7 @@ impl RoomSeats { pub fn admit_sidecar(&self, ptys: &PtyState, claim: &mcp_config::codex::admission::Claim) -> u16 { let mut seats = self.seats.lock(); let (launch, live, sidecar) = match seats.get_mut(&seat_key(&claim.room_id, &claim.account_id)) { - Some(Seat::Starting { requires_admission: true, sidecar: None }) => return 425, + Some(Seat::Starting { requires_admission: true, sidecar: None, .. }) => return 425, Some(Seat::Starting { requires_admission: false, .. }) => return 409, Some(Seat::Starting { sidecar: Some(starting), .. }) => { (starting.launch_id.clone(), starting.is_live(), &mut starting.sidecar) @@ -745,8 +748,34 @@ impl RoomSeats { } } + pub fn control_binding(&self, topic: &str, account: &str, supplied: &str, ptys: &PtyState) -> bool { + let seats = self.seats.lock(); + let (expected, live) = match seats.get(&seat_key(topic, account)) { + Some(Seat::Starting { generation, .. }) => (Some(generation.as_str()), true), + Some(Seat::Running(session)) => (Some(session.generation.as_str()), ptys.is_running(&session.pty_id)), + None => (None, false), + }; + mcp_config::session_control::capability_matches(Some(supplied), expected, live) + } + + pub fn generation_of(&self, topic: &str, account: &str) -> Option { + match self.seats.lock().get(&seat_key(topic, account)) { + Some(Seat::Starting { generation, .. }) => Some(generation.clone()), + Some(Seat::Running(session)) => Some(session.generation.clone()), + _ => None, + } + } + + pub fn launch_id_of(&self, topic: &str, account: &str, pty: &str) -> Option { + match self.seats.lock().get(&seat_key(topic, account)) { + Some(Seat::Running(session)) if session.pty_id == pty => session.launch_id.clone(), + _ => None, + } + } + pub fn new() -> Self { RoomSeats { + lifecycle: Arc::new(Mutex::new(())), seats: Arc::new(Mutex::new(BTreeMap::new())), } } @@ -792,7 +821,7 @@ impl RoomSeats { if taken { return Err(()); } - seats.insert(key, Seat::Starting { requires_admission: mcp_config::codex::admission::required_for(cli), sidecar: None }); + seats.insert(key, Seat::Starting { generation: uuid::Uuid::new_v4().to_string(), requires_admission: mcp_config::codex::admission::required_for(cli), sidecar: None }); Ok(()) } @@ -826,9 +855,12 @@ impl RoomSeats { fn hold(&self, account_id: &str, mut session: RunningSession) { let key = seat_key(&session.topic_id, account_id); let mut seats = self.seats.lock(); - if let Some(Seat::Starting { sidecar: Some(starting), .. }) = seats.remove(&key) { - session.sidecar = mcp_config::codex::admission::Admission::inherit( - starting.sidecar, &starting.launch_id, session.launch_id.as_deref()); + if let Some(Seat::Starting { generation, sidecar, .. }) = seats.remove(&key) { + session.generation = generation; + if let Some(starting) = sidecar { + session.sidecar = mcp_config::codex::admission::Admission::inherit( + starting.sidecar, &starting.launch_id, session.launch_id.as_deref()); + } } seats.insert(key, Seat::Running(session)); } @@ -1206,14 +1238,23 @@ fn codex_home(user_home: &Path, account_env: &[(String, String)], cwd: &Path) -> /// (#156, 決定6). A kind naming a CLI holds the way back into one of its /// sessions, so that is the line; a kind naming none has only the field, which /// is the one the person writes and the one the form shows for that kind. +#[derive(Clone)] +pub(crate) enum LaunchPolicy { Ui, Fresh, Resume(Option) } + fn resolve_launch( app: &AppHandle, account: &Account, topic: &TopicRef, cwd: &Path, + policy: &LaunchPolicy, ) -> Result { let cli = account.kind.cli(); - let recorded = room_log::session_of(app, &topic.topic_id, &account.id); + let recorded = if matches!(policy, LaunchPolicy::Fresh) { None } else { room_log::session_of(app, &topic.topic_id, &account.id) }; + if let LaunchPolicy::Resume(expected) = policy { + if recorded.is_none() || expected.as_ref().is_some_and(|id| recorded.as_ref() != Some(id)) { + return Err("No matching recorded conversation to resume; start fresh explicitly if intended".into()); + } + } let resume = match cli { Some(cli) => cli.resume_command(), None => account @@ -1223,6 +1264,10 @@ fn resolve_launch( .filter(|line| !line.is_empty()), }; + if matches!(policy, LaunchPolicy::Resume(_)) && resume.is_none() { return Err("This account has no resume command".into()); } + if matches!(policy, LaunchPolicy::Resume(_)) && cli.is_none() && !resume.is_some_and(|line| declares_session_id(&split_launch_options(line))) { + return Err("The configured resume command must carry {session_id}; no fresh conversation started".into()); + } let mut dropped_resume = None; if let (Some(session_id), Some(line)) = (recorded.as_deref(), resume) { if transcript_found(app, account, cli, cwd, session_id) { @@ -1269,6 +1314,9 @@ fn resolve_launch( }); } + if matches!(policy, LaunchPolicy::Resume(_)) { + return Err(format!("Conversation {session_id} cannot be resumed: transcript is missing; no fresh conversation started")); + } // The id is named, so what goes is this record and not one a relaunch // wrote in between (`room_log::forget_session`). Answering false is // that check having held, not a failure, and nothing is claimed in that @@ -1395,6 +1443,24 @@ pub fn start_session( cols: u16, rows: u16, ) -> Result { + let lifecycle = seats.lifecycle.clone(); + let _guard = lifecycle.lock(); + start_session_with_policy(app, room, pty_state, seats, account, topic_id, cols, rows, LaunchPolicy::Ui) +} + +pub(crate) fn preflight_resume(app: &AppHandle, account: &Account, topic: &TopicRef) -> Result { + let cwd = PathBuf::from(account.cwd.as_deref().unwrap_or_default()); + if !cwd.is_dir() { return Err("Account working directory is missing".into()); } + let line = resolve_launch(app, account, topic, &cwd, &LaunchPolicy::Resume(None))?; + line.resumed_from.ok_or_else(|| "No conversation to resume".into()) +} + +pub(crate) fn start_session_with_policy( + app: AppHandle, room: tauri::State, pty_state: tauri::State, + seats: tauri::State, account: Account, topic_id: String, cols: u16, rows: u16, + policy: LaunchPolicy, +) -> Result { + if app.state::().is_closing() { return Err("Pullcept is closing".into()); } let name = account.address_name().to_string(); if name.is_empty() { return Err( @@ -1464,7 +1530,7 @@ pub fn start_session( // Resume or fresh, decided here so every check below runs against the line // that will actually be spawned. Checking the account's own options and // then spawning the resume line would be checking the wrong line. - let launch_line = resolve_launch(&app, &account, &topic, &cwd)?; + let launch_line = resolve_launch(&app, &account, &topic, &cwd, &policy)?; // Whether this seat is being taken in front of posts it does not have. // @@ -1730,10 +1796,11 @@ pub fn start_session( seats.hold( &account.id, RunningSession { + generation: String::new(), launch_id: started.launch_id.clone(), sidecar: Default::default(), native_id: started.native_id.clone().or_else(|| launch_line.resumed_from.clone()) - .or_else(|| if account.kind == AccountKind::ClaudeCode { launch_line.session_id.clone() } else { None }), + .or_else(|| if matches!(account.kind, AccountKind::ClaudeCode | AccountKind::Cli) { launch_line.session_id.clone() } else { None }), pty_id: started.pty_id.clone(), started_at: started.started_at.clone(), // The line that ran, which on a resume is not the account's @@ -1945,6 +2012,7 @@ fn launch( // shared `.mcp.json` after another launch had added an entry its // `disabledMcpjsonServers` could not name starts that entry's sidecar too, // and without this it would take a seat under the other account's id. + env.push((mcp_config::session_control::CAPABILITY_ENV, app.state::().generation_of(topic_id, &account.id).ok_or("Launch control capability unavailable")?)); env.push((LAUNCHED_AS_ENV, account.id.clone())); env.push(("PULLCEPT_ACCOUNT_NAME", account.name.trim().to_string())); // The topic this CLI is launched into, beside the account (#264). The @@ -2063,6 +2131,7 @@ fn launch_codex_app_server( .collect(); server_env.extend( [ + (mcp_config::session_control::CAPABILITY_ENV, app.state::().generation_of(topic_id, &account.id).ok_or("Launch control capability unavailable")?), (mcp_config::codex::LAUNCH_ID_ENV, launch_id.clone()), (mcp_config::codex::admission::REQUIRED_ENV, "1".into()), (ROOM_TOKEN_ENV, room.token()), diff --git a/src-tauri/src/session_control.rs b/src-tauri/src/session_control.rs new file mode 100644 index 0000000..ca1279b --- /dev/null +++ b/src-tauri/src/session_control.rs @@ -0,0 +1,408 @@ +//! Authenticated room participants request fixed account operations; the app owns execution. +use crate::{ + config::{self, AccountKind}, + pty::PtyState, + room::{self, RoomState}, + session::{self, LaunchPolicy, RoomSeats}, +}; +use mcp_config::session_control::{Book, Connection, Mode, Operation, Request}; +use parking_lot::Mutex; +use serde_json::{json, Value}; +use std::{ + collections::BTreeMap, + path::PathBuf, + sync::atomic::{AtomicBool, Ordering}, +}; +use tauri::{AppHandle, Emitter, Manager}; + +#[derive(Default)] +struct Journal { + path: Option, + book: Book, +} +#[derive(Default)] +pub struct SessionControl { + journal: Mutex, + connections: Mutex>, + closing: AtomicBool, +} + +fn save_book(path: Option<&PathBuf>, book: &Book) -> Result<(), String> { + let path = path.ok_or("Session operations journal is unavailable")?; + let bytes = serde_json::to_vec_pretty(book).map_err(|e| e.to_string())?; + let temporary = path.with_extension("tmp"); + use std::io::Write; + let mut file = std::fs::File::create(&temporary) + .map_err(|e| format!("Cannot write session operations: {e}"))?; + file.write_all(&bytes) + .and_then(|_| file.sync_all()) + .map_err(|e| format!("Cannot flush session operations: {e}"))?; + drop(file); + std::fs::rename(&temporary, path).map_err(|e| format!("Cannot save session operations: {e}")) +} + +fn save(journal: &Journal) -> Result<(), String> { + save_book(journal.path.as_ref(), &journal.book) +} + +impl SessionControl { + pub fn initialize(&self, app: &AppHandle) -> Result<(), String> { + let root = app.path().app_data_dir().map_err(|e| e.to_string())?; + std::fs::create_dir_all(&root).map_err(|e| e.to_string())?; + let path = root.join("session-operations.json"); + let mut book = match std::fs::read(&path) { + Ok(bytes) => serde_json::from_slice::(&bytes) + .map_err(|e| format!("Cannot read session operations: {e}"))?, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Book::default(), + Err(e) => return Err(e.to_string()), + }; + book.interrupt(&room::now_iso()); + let journal = Journal { + path: Some(path), + book, + }; + save(&journal)?; + *self.journal.lock() = journal; + Ok(()) + } + pub fn is_closing(&self) -> bool { + self.closing.load(Ordering::SeqCst) + } + pub fn close(&self, app: &AppHandle) { + self.closing.store(true, Ordering::SeqCst); + let mut journal = self.journal.lock(); + journal.book.interrupt(&room::now_iso()); + if let Err(error) = save(&journal) { + crate::room_log::report(app, error); + } + } + pub fn connected( + &self, + origin: &str, + socket: &str, + topic: &str, + account: &str, + capability: &str, + launch: Option<&str>, + ) { + self.connections.lock().insert( + origin.into(), + Connection { + socket: socket.into(), + topic: topic.into(), + account: account.into(), + generation: Some(capability.to_owned()), + launch: launch.map(str::to_owned), + }, + ); + } + pub fn disconnected(&self, origin: &str, socket: &str) { + let mut connections = self.connections.lock(); + if connections.get(origin).is_some_and(|c| c.socket == socket) { + connections.remove(origin); + } + } + fn matches(&self, app: &AppHandle, connection: &Connection, pty: &str) -> bool { + let seats = app.state::(); + let live = seats + .running_pty(&connection.topic, &connection.account) + .as_deref() + == Some(pty) + && app.state::().is_running(pty); + connection.matches_live( + live, + seats + .generation_of(&connection.topic, &connection.account) + .as_deref(), + seats + .launch_id_of(&connection.topic, &connection.account, pty) + .as_deref(), + ) + } + fn room_connected(&self, app: &AppHandle, topic: &str, account: &str, pty: &str) -> bool { + self.connections + .lock() + .values() + .any(|c| c.topic == topic && c.account == account && self.matches(app, c, pty)) + } + /// Also checked by the socket's launch admission before dispatch. This check + /// excludes unaffiliated clients and sockets from a seat that already ended. + pub fn authorize(&self, app: &AppHandle, origin: &str, socket: &str) -> Result<(), String> { + let connections = self.connections.lock(); + let connection = connections + .get(origin) + .filter(|c| c.socket == socket) + .ok_or("Only a connected account session may control sessions")?; + let pty = app + .state::() + .running_pty(&connection.topic, &connection.account) + .ok_or("Requester's session is not running")?; + connection.authorize( + socket, + app.state::().is_running(&pty), + app.state::() + .generation_of(&connection.topic, &connection.account) + .as_deref(), + app.state::() + .launch_id_of(&connection.topic, &connection.account, &pty) + .as_deref(), + ) + } + fn operation_value(&self, app: &AppHandle, operation: &Operation) -> Value { + let mut result = serde_json::to_value(operation).unwrap_or(Value::Null); + let pty = operation.new_pty_id.as_deref(); + result["process_running"] = + json!(pty.is_some_and(|id| app.state::().is_running(id))); + result["room_connected"] = json!(pty.is_some_and(|id| self.room_connected( + app, + &operation.request.topic_id, + &operation.request.account_id, + id + ))); + result + } + pub fn status(&self, app: &AppHandle, id: &str) -> Result { + let op = self + .journal + .lock() + .book + .operations + .get(id) + .cloned() + .ok_or("Unknown operation_id")?; + Ok(self.operation_value(app, &op)) + } + pub fn list(&self, app: &AppHandle) -> Result { + let config = config::read_config(app)?; + let accounts: Vec<_> = config + .accounts + .iter() + .map(|a| { + json!({ + "account_id": a.id, "name": a.address_name(), "kind": a.kind, + "launchable": !matches!(a.kind, AccountKind::Admin | AccountKind::Mcp), + }) + }) + .collect(); + let seats: Vec<_> = app.state::().seated(&app.state::()).iter().map(|seat| { + let running = seat.session.as_ref(); + json!({ "topic_id": seat.topic_id, "account_id": seat.account_id, + "state": if running.is_some() { "running" } else { "starting" }, + "pty_id": running.map(|s| &s.pty_id), "started_at": running.map(|s| &s.started_at), + "resumed_from": running.and_then(|s| s.resumed_from.as_deref()), + "room_connected": running.is_some_and(|s| self.room_connected(app, &seat.topic_id, &seat.account_id, &s.pty_id)), + }) + }).collect(); + Ok( + json!({ "accounts": accounts, "topics": app.state::().open_topics(), "seats": seats }), + ) + } + pub fn accept( + &self, + app: &AppHandle, + request: Request, + account: &str, + topic: &str, + name: &str, + ) -> Result { + if self.is_closing() { + return Err("Pullcept is closing".into()); + } + let id = request.operation_id.clone(); + let is_new = { + let mut journal = self.journal.lock(); + let path = journal.path.clone(); + journal.book.accept_durable( + request, + account, + topic, + name, + &room::now_iso(), + |book| save_book(path.as_ref(), book), + )? + }; + if is_new { + self.notice(app, &id); + let worker_app = app.clone(); + let worker_id = id.clone(); + // Owned by the app runtime, never by serve_participant or a sidecar. + tauri::async_runtime::spawn_blocking(move || { + let state = worker_app.state::(); + state.execute(&worker_app, &worker_id); + }); + } + self.status(app, &id) + } + fn update( + &self, + app: &AppHandle, + id: &str, + status: &str, + pty: Option, + error: Option, + ) -> Result<(), String> { + let mut journal = self.journal.lock(); + let op = journal + .book + .operations + .get_mut(id) + .ok_or("Unknown operation_id")?; + if self.is_closing() { + return Err("Pullcept is closing; operation interrupted".into()); + } + op.status = status.into(); + op.updated_at = room::now_iso(); + op.new_pty_id = pty; + op.error = error; + if let Err(err) = save(&journal) { + let op = journal + .book + .operations + .get_mut(id) + .ok_or("Unknown operation_id")?; + if status == "running" { + op.status = "failed".into(); + } + op.error = Some(format!("Operation record save failed: {err}")); + crate::room_log::report(app, err.clone()); + return Err(err); + } + Ok(()) + } + fn notice(&self, app: &AppHandle, id: &str) { + let Some(op) = self.journal.lock().book.operations.get(id).cloned() else { + return; + }; + let content = format!("Session operation {id}: requester {} ({}, topic {}), target {}/{} {:?} expected {:?}; {} new_pty {:?}{}", + op.requester_name, op.requester_account, op.requester_topic, op.request.topic_id, op.request.account_id, + op.request.action, op.request.expected_pty_id, op.status, op.new_pty_id, op.error.as_ref().map(|e| format!(": {e}")).unwrap_or_default()); + room::post_app_notice( + app, + &app.state::(), + &op.requester_topic, + &content, + None, + ); + let _ = app.emit("session-operation", self.operation_value(app, &op)); + } + fn execute(&self, app: &AppHandle, id: &str) { + let seats = app.state::(); + let _guard = seats.lifecycle.lock(); + if self.update(app, id, "running", None, None).is_err() { + self.notice(app, id); + return; + } + let Some(op) = self.journal.lock().book.operations.get(id).cloned() else { + return; + }; + let result = self.run(app, &op.request); + let (status, pty, error) = match result { + Ok(Some(pty)) => ("process_started", Some(pty), None), + Ok(None) => ("stopped", None, None), + Err(err) => ("failed", None, Some(err)), + }; + if let Err(err) = self.update(app, id, status, pty, error) { + crate::room_log::report(app, err); + } + self.notice(app, id); + let _ = app.emit("session-seats-changed", ()); + } + fn run(&self, app: &AppHandle, request: &Request) -> Result, String> { + if self.is_closing() { + return Err("Pullcept is closing".into()); + } + let account = config::read_config(app)? + .accounts + .into_iter() + .find(|a| a.id == request.account_id) + .ok_or("Unknown configured account")?; + if matches!(account.kind, AccountKind::Admin | AccountKind::Mcp) { + return Err("Target account is not a launchable session".into()); + } + mcp_config::session_control::execute(request, &mut AppDriver { app, account }) + } +} + +struct AppDriver<'a> { + app: &'a AppHandle, + account: config::Account, +} +impl mcp_config::session_control::Driver for AppDriver<'_> { + fn current( + &mut self, + request: &Request, + ) -> Result, String> { + if self + .app + .state::() + .topic_of(&request.topic_id) + .is_none() + { + return Err("Target topic is not open".into()); + } + let seats = self.app.state::(); + let ptys = self.app.state::(); + let held = seats.seated(&ptys); + let seat = held + .iter() + .find(|s| s.topic_id == request.topic_id && s.account_id == request.account_id); + if seat.is_some_and(|s| s.session.is_none()) { + return Err("Target seat is starting; wait for its PTY before controlling it".into()); + } + Ok(seat.map(|seat| mcp_config::session_control::Current { + pty_id: seat + .session + .as_ref() + .map(|s| s.pty_id.clone()) + .unwrap_or("starting".into()), + native_id: seat + .session + .as_ref() + .and_then(|s| seats.native_id_of(&request.topic_id, &request.account_id, &s.pty_id)) + .flatten(), + resumed_from: seat.session.as_ref().and_then(|s| s.resumed_from.clone()), + })) + } + fn resume(&mut self, request: &Request) -> Result { + let topic = self + .app + .state::() + .topic_of(&request.topic_id) + .ok_or("Target topic is not open")?; + session::preflight_resume(self.app, &self.account, &topic) + } + fn stop(&mut self, request: &Request) -> Result<(), String> { + let id = request + .expected_pty_id + .as_deref() + .ok_or("Expected PTY required")?; + self.app.state::().kill_exact(id)?; + crate::codex_app_server::stop_one(self.app, id); + Ok(()) + } + fn start( + &mut self, + request: &Request, + mode: Mode, + resume_id: Option<&str>, + ) -> Result { + let policy = match mode { + Mode::Fresh => LaunchPolicy::Fresh, + Mode::Resume => { + LaunchPolicy::Resume(Some(resume_id.ok_or("Resume ID required")?.into())) + } + }; + let started = session::start_session_with_policy( + self.app.clone(), + self.app.state::(), + self.app.state::(), + self.app.state::(), + self.account.clone(), + request.topic_id.clone(), + 100, + 30, + policy, + )?; + Ok(started.pty_id) + } +} diff --git a/src/main.ts b/src/main.ts index 05ffd0d..3c3aa3b 100644 --- a/src/main.ts +++ b/src/main.ts @@ -3852,6 +3852,10 @@ async function main(): Promise { // (#82). A report arrives on every assistant message, and the two that cross // the threshold are the only ones the row has anything to say about — the // rest move the facts column alone. + await listen("session-seats-changed", () => void refreshSeats()); + await listen<{ status: string; error?: string }>("session-operation", (event) => { + if (event.payload.error) status(`セッション操作 ${event.payload.status}: ${event.payload.error}`, "error"); + }); await listen("session-stats", (event) => sessions.receiveStats(event.payload)); // A Codex app-server seat (#326) or a Claude Code seat's hooks (#331) said // what it is doing. Keyed on the seat @@ -4079,7 +4083,7 @@ async function main(): Promise { // and a room that fails to answer is no reason to leave it unreachable — the // failure this is here for is the screen having been reloaded, which the room // knows nothing about (#84). It swallows its own errors, so it needs none. - await refreshSeats(); + await refreshSeats(true); try { await join(); diff --git a/src/session-controller.ts b/src/session-controller.ts index 5f4e3a6..f0050e8 100644 --- a/src/session-controller.ts +++ b/src/session-controller.ts @@ -85,7 +85,7 @@ export function createSessionController(deps: Dependencies) { * that otherwise belongs entirely to the session (#84). */ const RESUMED_NOTICE = - "\x1b[2m[pullcept] 画面が再読み込みされました。セッションは走ったままで、この端末はそこへ繋ぎ直したものです。これより前の出力は残っていません。\x1b[0m"; + "\x1b[2m[pullcept] 稼働中のセッションに接続しました。端末への接続より前の出力は復元されません。\x1b[0m"; /** * How long a terminal must stay silent before its row stops saying that it is @@ -443,7 +443,7 @@ export function createSessionController(deps: Dependencies) { * after a launch and after a session exits, which are the two moments the * answer changes. */ - async function refreshSeats(): Promise { + async function refreshSeats(restoreDiagnostics = false): Promise { try { const held = await deps.invoke("seated_accounts"); seated = new Map(held.map((seat) => [deps.seatKey(seat.topic_id, seat.account_id), seat])); @@ -452,7 +452,7 @@ export function createSessionController(deps: Dependencies) { // on a failed read. It still redraws: what failed is this one value, and // whatever else moved since the last draw is not held back by it. } - await adoptSeats(); + await adoptSeats(restoreDiagnostics); deps.renderPanel(); // The list marks the topics holding a seat, and this is the answer that // changed (#141, AI 判断5). @@ -476,16 +476,21 @@ export function createSessionController(deps: Dependencies) { * a session that has printed nothing, and reading what a session last printed * is how the person decides whether to end it (#57). * - * Only ever after a reload, never after a restart: the seats are the app's own - * memory and go with it, so an app that has just started holds none. + * Also used after app-owned MCP launches and restarts. Their new PTY replaces + * an older view for the same seat; app restart itself still holds no live seats. */ - async function adoptSeats(): Promise { + async function adoptSeats(restoreDiagnostics: boolean): Promise { const adopted: string[] = []; for (const seat of seated.values()) { // No session yet: the seat is claimed and the launch is still in flight. - // Nothing to subscribe to, and it is this screen's own launch in every case - // but a reload landing inside that window. - if (!seat.session || views.has(deps.seatKey(seat.topic_id, seat.account_id))) continue; + // Nothing to subscribe to until either a UI or MCP launch has a PTY. + if (!seat.session) continue; + const existing = views.get(deps.seatKey(seat.topic_id, seat.account_id)); + if (existing) { + // Same process already adopted or the UI's own launch still attaching. + if (!existing.ptyId || existing.ptyId === seat.session.pty_id) continue; + discardView(existing); + } const account = deps.accounts.find((one) => one.id === seat.account_id); // An account this screen does not have is one it cannot draw a row for, and // the row is the only way that terminal could be reached. The app refuses @@ -497,10 +502,8 @@ export function createSessionController(deps: Dependencies) { adopted.push(viewName(view)); } if (!adopted.length) return; - // Open, because open is where it was: a session is reloaded out from under - // while it is being watched, which is to say while this pane is showing it. - deps.revealDiagnostics(); - deps.status(`${adopted.join("、")} の端末に繋ぎ直しました。再読み込みより前の出力は残っていません。`); + if (restoreDiagnostics) deps.revealDiagnostics(); + deps.status(`${adopted.join("、")} の稼働中の端末に接続しました。接続より前の出力は復元されません。`); } /**