diff --git a/docs/internals/architecture/appserver/app-protocol-v1.schema.json b/docs/internals/architecture/appserver/app-protocol-v1.schema.json index 9e3db24cb..2368932e5 100644 --- a/docs/internals/architecture/appserver/app-protocol-v1.schema.json +++ b/docs/internals/architecture/appserver/app-protocol-v1.schema.json @@ -34,6 +34,8 @@ "mux/member/open", "mux/member/close", "session/snapshot", + "session/models", + "session/model/select", "turn/start", "turn/steer", "turn/follow_up", diff --git a/gui/README.md b/gui/README.md index db3b4b507..6d2b733db 100644 --- a/gui/README.md +++ b/gui/README.md @@ -3,7 +3,7 @@ This directory contains the Product-neutral native HarnessGUI shell accepted by [GUI-B0 issue #582](https://github.com/zhnt/loushang/issues/582). React and TypeScript own presentation and UI-local state; Tauri/Rust owns desktop -integration and adapts the accepted read-only App Contract slice. The GUI does not +integration and adapts the accepted App Contract slice. The GUI does not own Product, Harness, AppService, or AppHost runtimes. The default B1 launch is deliberately offline. It provides a Product-neutral UI @@ -12,11 +12,12 @@ Sessions, isolated per-Session drafts, active Run and Task progress, expandable Activity details, distinct root/subagent AgentRuns, and read-only ChangeSet review through a central Quick Look and the Work Dock. Every sample value is labelled as fixture data. It does not read a repository, start a Python backend, -or request a model. An opt-in B2 Windows launch can attach read-only to an +or request a model. An opt-in B2 Windows launch can attach to an existing AppHost, install one barriered initial Session snapshot, publish validated ongoing event rounds with authoritative replacement snapshots, and -reconnect through a fresh snapshot after transport loss or an event gap. It -does not enable any mutation. Further cross-language and real-service work +reconnect through a fresh snapshot after transport loss or an event gap. It can +submit text and interrupt the current execution in that existing Session over +the same authenticated controller connection. Further cross-language and real-service work remains in C1/B2 as described in the [engineering plan](../docs/internals/architecture/drafts/gui-engineering-bootstrap-plan.md). @@ -57,7 +58,7 @@ executable without producing installers. `dev:fixture-native` and `build:fixture-native` enable the B1-only Rust invoke/event canary; the command is absent from the default native build. -## Read-only live launch +## Controlled live launch The AppHost/desktop launcher may start the built executable with both native arguments below. Omitting both keeps the offline fixture. Partial, duplicate or @@ -72,14 +73,17 @@ The record root and Mux selection are parsed and retained by Rust; the WebView cannot provide or read them. Native admission still validates the private record directory and file before authentication. The published payload contains no attachment ID, controller generation or authentication key. Live snapshots are -visibly labelled read-only; fixture playback, Send, Interrupt, Workspace, -ChangeSet, Tasks and Subagents remain unavailable until their own accepted live -contracts are implemented. Closing the window cooperatively detaches and joins +visibly labelled as AppHost-controlled. Send and Interrupt use controller facts +kept only in Rust; fixture playback, Workspace, ChangeSet, Tasks and Subagents +remain unavailable until their own accepted live contracts are implemented. Closing the window cooperatively detaches and joins the reader without stopping the shared AppHost. Each connection attempt has a new epoch. A failed transport or invalid round marks live facts disconnected/resync-required; the native adapter rereads the private record, authenticates and attaches again, and the WebView resumes only after a complete -fresh snapshot and membership barrier. Unknown mutations are never replayed. +fresh snapshot and membership barrier. A submit with an uncertain response is +looked up by its stable submission ID after reconnect and is never replayed +automatically. The draft is cleared only after acceptance is known; commands are +blocked until the resulting authoritative event or snapshot advances the Session. For an opt-in desktop acceptance run, first build the release executable and then launch an isolated real AppHost plus two real mux/session projections: @@ -92,19 +96,34 @@ pnpm --dir gui run accept:live-native The second command opens a visible HarnessGUI window and keeps the shared AppHost alive until that window is closed. It also rotates the real local listener and connection record, and waits for the GUI to reconnect and accept a -fresh authoritative snapshot. Inspect the final connected/read-only labels, -real Session projection, disabled mutation controls and normal/maximized layout, +fresh authoritative snapshot. Inspect the final connected labels, real Session +projection, Send/Interrupt behavior and normal/maximized layout, then close the window. The launcher verifies the reconnect, that the GUI released its mux controller, that the AppHost is still accepting, and that an independent peer Session is still snapshot-readable. Non-secret lifecycle evidence is written under `gui/test-results/live-apphost-desktop/`; this opt-in helper is not part of the rapid GUI gate and does not invoke a model. +To exercise the same isolated AppHost with the saved Coding default model and +its existing credential, run the explicit provider acceptance instead: + +```text +pnpm --dir gui run accept:live-provider +``` + +This command resolves the model through the normal layered AI ModelRegistry and +Coding settings; the GUI does not parse `models.json` or receive credentials. +It currently uses no tools, sends one short acceptance prompt, may incur provider +usage, and then leaves the window open for manual conversation. Non-secret +evidence is written under `gui/test-results/live-provider-desktop/`. It is never +part of `check`, CI or another default gate. + The first observed live desktop run is recorded in [Windows live AppHost acceptance](tests/playback/windows-live-apphost-acceptance.md). -It records the initial read-only projection, ongoing event integration, -disconnect/resynchronization and cooperative desktop-exit lifecycle. Mutation -and Windows display scaling remain outside this slice. +It records the initial projection, ongoing event integration, +disconnect/resynchronization and cooperative desktop-exit lifecycle. The +automated integration check additionally submits through the production desktop +control bridge. Windows display scaling remains outside this slice. For a standalone offline executable, use `pnpm --dir gui run build:fixture-native` from the repository root. Do not substitute a plain `cargo build`: that bypasses diff --git a/gui/contracts/README.md b/gui/contracts/README.md index 64a9abc51..6ede0449e 100644 --- a/gui/contracts/README.md +++ b/gui/contracts/README.md @@ -1,7 +1,7 @@ # C1 execution contract probes -Status: **partial C1 contract coverage plus the first B2 read-only native adapter**; -not an accepted full codec or mutable client. +Status: **partial C1 contract coverage plus a bounded B2 existing-Session control adapter**; +not an accepted full codec or general mutable client. The [contract generation and native connection plan](connection-plan.md) now records the selected reference inputs, generated candidate DTOs and the exact @@ -19,7 +19,7 @@ the value/selection contract and its use by the native-file and authentication probes. A subsequent [Windows loopback lifecycle experiment](connection-lifecycle-evidence.md) now composes record admission, authentication, hello, deadlines and cancellation. The bounded connection implementation is now shared with the Tauri host. -The [read-only attachment experiment](attachment-lifecycle-evidence.md) now +The [attachment experiment](attachment-lifecycle-evidence.md) now extends the loopback probe through member snapshots and detach; real AppHost snapshot publication is verified below, while native-window acceptance remains. @@ -29,7 +29,7 @@ The four local probe slices are one offline compatibility work item, not four production integration milestones. Their scope is measured against [bootstrap plan section 6](../../docs/internals/architecture/drafts/gui-engineering-bootstrap-plan.md#6-appclient-跨语言接入与生命周期). The default GUI still uses fixtures. A Windows launch with both admitted native -arguments selects the read-only live client; protocol validators now live under +arguments selects the controlled live client; protocol validators now live under `src-tauri` and the probes consume that same implementation. | Obligation | Evidence retained | Remaining work before claiming coverage | @@ -56,9 +56,10 @@ assertions; do not expand case counts merely to suggest readiness. obligations required by the first connection. Add reverse encoding evidence, rather than relying on Python-to-Rust-to-TypeScript projection alone. 3. The bounded B2 connection slice now reaches explicit profile negotiation, - authenticated attachment, a membership-barriered initial snapshot and owned - cleanup. Keep submit/recovery acceptance separate; live Send and Interrupt - remain disabled. + authenticated attachment, a membership-barriered initial snapshot, ongoing + event publication, submit/interrupt control and owned cleanup. Control stays + on the attachment-owning connection. An uncertain submit is queried by stable + submission ID after reconnect and is never sent a second time automatically. Workspace/ChangeSet integration remains pending its own accepted contracts; execution connectivity must not silently turn fixture repository data into live @@ -212,7 +213,7 @@ Reproduce this offline evidence with `pnpm --dir gui run check:contract`. ## Real AppHost integration follow-up -The first real AppHost/AppService read-only integration and its request-ID fix +The real AppHost/AppService integration, desktop control path and request-ID fix are documented in [AppHost integration evidence](apphost-integration-evidence.md). Run `uv run python scripts/gui/check_apphost_integration.py` explicitly on Windows; -this does not enable live GUI publication or add a default gate. +this remains an opt-in GUI integration check and adds no default gate. diff --git a/gui/contracts/apphost-integration-evidence.md b/gui/contracts/apphost-integration-evidence.md index 2ff76cf8e..05bcd5ea3 100644 --- a/gui/contracts/apphost-integration-evidence.md +++ b/gui/contracts/apphost-integration-evidence.md @@ -1,6 +1,7 @@ -# Real AppHost read-only integration +# Real AppHost integration -Status: B2 read-only event and resynchronization evidence, updated 2026-09-22. +Status: B2 event, resynchronization and existing-Session control evidence, +updated 2026-09-22. Run from the repository root on Windows: @@ -11,8 +12,9 @@ uv run python scripts/gui/check_apphost_integration.py This opt-in script constructs the real Coding continuity attempt, AppHost catalog/runtime, AppService, HostedLocalRuntimeV1, authenticated LocalAppServer, and real Coding session factory. It uses temporary private connection records, -continuity and session stores. Installed-code admission pins are test-owned; -model streaming is explicitly forbidden and no work is submitted. It does not +continuity and session stores. Installed-code admission pins are test-owned. +The control case uses a deterministic process-local model transport and submits +one prompt; it makes no network provider call. It does not connect to an existing user application, read user settings, or start a TUI. The existing Rust connection probe authenticates to that actual server and @@ -35,6 +37,15 @@ ordered member/Session identity and one execution snapshot against the real AppHost. The serialized publication contains neither `attachmentId` nor `controllerGeneration`; those controller capabilities remain native. +The `control_probe` then uses that same production Rust adapter to attach and +submit on one authenticated connection. AppHost routes the prompt into the real +Coding `AgentSession`; the deterministic assistant stream returns through the +normal execution event and snapshot path. The probe verifies acceptance plus +nonempty event publication. This caught three integration errors that isolated +codecs could not: controller authority is connection-bound, controller +generation is numeric on the wire, and request IDs increase across snapshot, +control and event operations on the connection. + ## Integration defect found The isolated probe previously sent textual request IDs (`attach`, `snapshot-0`, @@ -50,8 +61,9 @@ No server contract or runtime implementation was changed to accommodate GUI. ## Limits and next slice -- The native adapter and connected-window lifecycle are accepted for this - read-only slice; display scaling and mutation remain separate work. +- The native adapter, connected-window lifecycle and bounded existing-Session + submit/interrupt path are covered; display scaling and broader mutation remain + separate work. - It verifies idle snapshots and empty event batches, real controller arbitration, graceful client EOF cleanup and explicit deployment settlement. It does not prove hard process termination cleanup or every failed-snapshot/disconnect race in the real host. @@ -60,12 +72,25 @@ No server contract or runtime implementation was changed to accommodate GUI. detach against the real AppHost. The desktop uses that same reader to publish the barriered initial snapshot and complete validated event rounds. On a transport failure or rejected round it establishes a fresh connection epoch - and replaces state from a new authoritative snapshot. Submission, approval - and takeover are not added. -- Model-free real Coding sessions do not establish live provider execution. + and replaces state from a new authoritative snapshot. Approval and takeover + are not added. Submit is never automatically replayed; after a response loss, + the adapter queries the stable submission ID on a new attachment. +- The deterministic Coding session proves AgentSession routing and streaming, + not authentication to or behavior of a network AI provider. - Nonempty contiguous event batches and invalid batches are covered by the scripted fixture; metadata/content watermark separation is additionally unit tested in Rust. - This remains an opt-in GUI integration check, not a TUI/Harness default gate. -Next add accepted live mutation contracts independently. Keep live facts -separate from offline fixtures and never infer mutation results across reconnect. +Next record a completed provider-backed desktop run, then add the separate New +Session, approval and attachment contracts. Keep live facts separate from +offline fixtures and never infer mutation results across reconnect. + +## Opt-in provider desktop acceptance + +`pnpm --dir gui run accept:live-provider` uses the same isolated AppHost and +desktop control bridge but resolves the saved Coding `default_model` through the +standard layered ModelRegistry and settings path. Credentials remain in the AI +runtime; they are not serialized to Rust or the WebView. The helper disables +tools, sends one bounded acceptance prompt, and records only the selected model +identity and lifecycle facts under `gui/test-results/live-provider-desktop/`. +It is intentionally interactive, network-using and outside every default gate. diff --git a/gui/contracts/attachment-lifecycle-evidence.md b/gui/contracts/attachment-lifecycle-evidence.md index c97e1ab83..b1a1173d5 100644 --- a/gui/contracts/attachment-lifecycle-evidence.md +++ b/gui/contracts/attachment-lifecycle-evidence.md @@ -98,5 +98,6 @@ not infer that a consistent live snapshot can be published during concurrent membership changes from this experiment alone. The separate [real AppHost integration](apphost-integration-evidence.md) now tests -actual ownership and idle event reads. Next expose the read-only native port with correct -event/barrier and cleanup ownership. Keep fixture UI and live facts separate. +actual ownership, idle event reads and the desktop control path. The native port +now retains control authority on the connection that acquired it, publishes +complete event rounds and owns cleanup. Keep fixture UI and live facts separate. diff --git a/gui/contracts/connection-plan.md b/gui/contracts/connection-plan.md index 955e9418d..6fba67296 100644 --- a/gui/contracts/connection-plan.md +++ b/gui/contracts/connection-plan.md @@ -54,7 +54,8 @@ Use the existing shared AppHost application; do not start a second GUI service. | Negotiate | Send authenticated APP mode, verify expected execution profile/instance and exact hello, then echo hello; see `remote_client.py` | Hello is not controller ownership or UI readiness | | Attach | Explicit mux selector, `mux/attach`, retain attachment ID and lossless controller generation | Do not silently take over an already-controlled mux | | Initial snapshot | Validate attachment membership/identity and obtain each member's execution composite snapshot using the negotiated instance | Do not mix fixture values or partial member snapshots into live state | -| Publish | Atomically install validated member snapshots and cursor barriers; expose an explicit read-only connected state | Send/approval/retry remain unavailable in this first slice | +| Publish | Atomically install validated member snapshots and cursor barriers; expose an explicit connected state | Do not enable capabilities that were not negotiated | +| Control | Submit or interrupt through the attachment-owning connection with exact instance, member and generation facts | Never transfer controller authority to the WebView or replay an uncertain submit | | Detach | Use current attachment/generation to detach, stop readers and close the borrowed connection | Never send STOP mode or shut down the shared application | Authentication uses the existing G16 transport profile even when the semantic @@ -81,8 +82,9 @@ silently sharing one mux. barrier before publication. Do not synthesize missing events from UI history. - Keep a single writer and ongoing response reader with bounded pending work; control/cleanup requests must not wait behind a long-running request lock. -- Disconnect cannot automatically replay submit, approval or mutation. This - slice does not implement any of those operations. +- Disconnect cannot automatically replay submit, approval or mutation. The + existing-Session slice may query a stable submission ID after reconnect, but + only a confirmed missing result permits the user to decide whether to retry. ## Delivery checkpoints, not new default gates @@ -105,18 +107,20 @@ silently sharing one mux. [Local attempt fencing](connection-epoch-evidence.md) now guards probe state updates and cancellation. Bounded mux membership rechecks reject a changed revision or member set after snapshots and event rounds. Continuous detection - and reusable desktop ownership are implemented for the read-only slice. + and reusable desktop ownership are implemented for the live slice. The attachment reader is now a reusable `ReadSession` with separate attach, initialize, poll and detach operations. The evidence driver alone chooses two rounds; the session does not own a loop or socket. Each poll commits all member watermarks only after its membership recheck. Request IDs fail closed at the existing signed-63-bit connection limit instead of wrapping. -4. **Implemented read-only publication slice:** the Tauri host starts the shared +4. **Implemented publication and existing-Session control slice:** the Tauri host starts the shared adapter from native-only launch facts, publishes one atomic initial snapshot and complete ongoing event rounds, and reconnects with a fresh epoch and snapshot after transport loss or a rejected round. React distinguishes live - data from fixtures and disables mutation. Native-window lifecycle acceptance - covers listener/record rotation, reattachment and cooperative exit. + data from fixtures. Submit and interrupt are serialized onto that same owned + connection; a lost submit response is reconciled by stable submission ID + after a fresh attachment. Native-window lifecycle acceptance covers listener/ + record rotation, reattachment and cooperative exit. Transport and attachment must pass their own evidence before enabling the GUI connection control. No live connection readiness is claimed by checkpoint 1. @@ -167,12 +171,13 @@ are present; partial, duplicate or unknown native arguments fail closed. Neither fact comes from the WebView. After authentication, attachment, every execution snapshot and the mux membership barrier succeed, one redacted snapshot is handed through a native event. Attachment/controller authority is never serialized. -The connection then continues validated read-only polling until desktop -stop/exit; this slice deliberately does not publish those event batches. +The connection then continues validated polling until desktop stop/exit and +publishes complete event rounds followed by authoritative replacement snapshots. The React client validates service-instance and membership identity, installs -the snapshot atomically, exposes no Workspace/ChangeSet capabilities, and keeps -Send/Interrupt disabled. An invalid or timed-out handoff asks native ownership +the snapshot atomically, and exposes no Workspace/ChangeSet capabilities. Send +and Interrupt call native commands only while connected; they are blocked until +the resulting authoritative publication advances the Session. An invalid or timed-out handoff asks native ownership to stop and join. The default launch remains the offline fixture. This is still not native-window acceptance with a real host. Forced process termination and Tauri restart (whose exit prevention is ignored by Tauri) are diff --git a/gui/contracts/rust/src/bin/connection_probe.rs b/gui/contracts/rust/src/bin/connection_probe.rs index a0bde4607..c794521fa 100644 --- a/gui/contracts/rust/src/bin/connection_probe.rs +++ b/gui/contracts/rust/src/bin/connection_probe.rs @@ -1,5 +1,6 @@ //! Windows lifecycle evidence consuming the same native adapter as Tauri. #[cfg(windows)] +#[allow(dead_code, unused_imports)] #[path = "../../../../src-tauri/src/app_client/mod.rs"] mod app_client; #[cfg(windows)] @@ -38,6 +39,7 @@ fn main() { attach, mux_name: "gui-fixture", stop: watch.then_some(stop), + control_commands: None, }, &mut |_| Ok(()), ) diff --git a/gui/contracts/rust/src/bin/control_probe.rs b/gui/contracts/rust/src/bin/control_probe.rs new file mode 100644 index 000000000..799703fd0 --- /dev/null +++ b/gui/contracts/rust/src/bin/control_probe.rs @@ -0,0 +1,126 @@ +//! Submit one deterministic turn through the production desktop control path. +#[cfg(windows)] +#[allow(dead_code, unused_imports)] +#[path = "../../../../src-tauri/src/app_client/mod.rs"] +mod app_client; +#[cfg(windows)] +#[path = "../../../../src-tauri/src/read_stop.rs"] +mod read_stop; +#[cfg(windows)] +#[allow(dead_code)] +#[path = "../../../../src-tauri/src/read_worker.rs"] +mod read_worker; + +#[cfg(windows)] +fn main() { + use std::io::Write; + + let args: Vec<_> = std::env::args_os().collect(); + let outcome = (|| { + if args.len() != 7 { + return Err(()); + } + let timeout: u64 = args[2].to_str().ok_or(())?.parse().map_err(|_| ())?; + if !(1..=5000).contains(&timeout) { + return Err(()); + } + let root = std::path::PathBuf::from(&args[1]); + let mux_name = args[3].to_str().ok_or(())?.to_owned(); + let session_id = args[4].to_str().ok_or(())?.to_owned(); + let submission_id = args[5].to_str().ok_or(())?.to_owned(); + let text = args[6].to_str().ok_or(())?.to_owned(); + read_worker::ReadWorker::spawn(move |stop| { + let finish = stop.clone(); + let (commands, command_receiver) = std::sync::mpsc::channel(); + let pending: std::cell::RefCell< + Option>>, + > = std::cell::RefCell::new(None); + let mut submitted = false; + app_client::AppClient::default() + .run_stream( + app_client::ConnectionOptions { + root: &root, + timeout: std::time::Duration::from_millis(timeout), + cancel: None, + attach: true, + mux_name: &mux_name, + stop: Some(stop), + control_commands: Some(&command_receiver), + }, + 1, + &mut |_| { + let receive = pending.borrow_mut().take().ok_or(())?; + let result = receive + .recv_timeout(std::time::Duration::from_secs(6)) + .map_err(|_| ())??; + let (accepted, reason) = match result { + app_client::ControlResult::Accepted => (true, None), + app_client::ControlResult::Missing => { + (false, Some("missing".to_owned())) + } + app_client::ControlResult::Rejected(reason) => { + (false, Some(reason)) + } + }; + println!( + "{}", + serde_json::json!({"type":"control","accepted":accepted,"reason":reason}) + ); + std::io::stdout().flush().map_err(|_| ())?; + if accepted { Ok(()) } else { Err(()) } + }, + &mut |round| { + let terminal = + round.members.iter().flat_map(|member| &member.events).any( + |event| { + matches!( + event["source"]["kind"].as_str(), + Some("turn_completed") | Some("turn_interrupted") + ) + }, + ); + println!("{}", serde_json::json!({"type":"round","payload":round})); + std::io::stdout().flush().map_err(|_| ())?; + if terminal { + finish.request(); + } + Ok(()) + }, + &mut |authority| { + let Some(_) = authority else { + return Ok(()); + }; + if submitted { + return Err(()); + } + submitted = true; + let (reply, receive) = std::sync::mpsc::sync_channel(1); + commands + .send(app_client::ControlCommand::Submit { + session_id: session_id.clone(), + submission_id: submission_id.clone(), + text: text.clone(), + reply, + }) + .map_err(|_| ())?; + pending.replace(Some(receive)); + Ok(()) + }, + &mut || Ok(true), + ) + .map(|_| ()) + }) + .map_err(|_| ())? + .wait() + .map_err(|_| ()) + })(); + if outcome.is_err() { + eprintln!("control probe failed"); + std::process::exit(1); + } +} + +#[cfg(not(windows))] +fn main() { + std::process::exit(2); +} diff --git a/gui/contracts/rust/src/bin/event_probe.rs b/gui/contracts/rust/src/bin/event_probe.rs index 77a4b5498..303f83d9c 100644 --- a/gui/contracts/rust/src/bin/event_probe.rs +++ b/gui/contracts/rust/src/bin/event_probe.rs @@ -1,5 +1,6 @@ //! Observe one validated live event round through the desktop AppClient adapter. #[cfg(windows)] +#[allow(dead_code, unused_imports)] #[path = "../../../../src-tauri/src/app_client/mod.rs"] mod app_client; #[cfg(windows)] @@ -37,6 +38,7 @@ fn main() { attach: true, mux_name: &mux_name, stop: Some(stop), + control_commands: None, }, 1, &mut |snapshot| { @@ -60,6 +62,7 @@ fn main() { } Ok(()) }, + &mut |_| Ok(()), &mut || Ok(true), ) .map(|_| ()) diff --git a/gui/contracts/rust/src/bin/snapshot_probe.rs b/gui/contracts/rust/src/bin/snapshot_probe.rs index e9e164c20..978e3eb5c 100644 --- a/gui/contracts/rust/src/bin/snapshot_probe.rs +++ b/gui/contracts/rust/src/bin/snapshot_probe.rs @@ -1,5 +1,6 @@ //! Redacted initial-snapshot evidence consuming the desktop AppClient adapter. #[cfg(windows)] +#[allow(dead_code, unused_imports)] #[path = "../../../../src-tauri/src/app_client/mod.rs"] mod app_client; #[cfg(windows)] @@ -33,6 +34,7 @@ fn main() { attach: true, mux_name: &mux_name, stop: None, + control_commands: None, }, &mut |snapshot| publish.send(snapshot).map_err(|_| ()), ) diff --git a/gui/contracts/rust/src/main.rs b/gui/contracts/rust/src/main.rs index fb032c501..b6e015da8 100644 --- a/gui/contracts/rust/src/main.rs +++ b/gui/contracts/rust/src/main.rs @@ -2,6 +2,7 @@ use serde::{Deserialize, Serialize}; use serde_json::{json, value::RawValue, Value}; use std::io::{self, BufRead}; +#[allow(dead_code)] #[path = "../../../src-tauri/src/app_client/projection.rs"] mod projection; // Compile the generated candidate DTOs without treating them as wire decoders. diff --git a/gui/package.json b/gui/package.json index 8dd7593d3..f30dc3438 100644 --- a/gui/package.json +++ b/gui/package.json @@ -33,6 +33,7 @@ "build": "node ../scripts/gui/run.mjs build-native", "build:fixture-native": "node ../scripts/gui/run.mjs build-fixture-native", "accept:live-native": "uv run python ../scripts/gui/run_live_apphost_acceptance.py", + "accept:live-provider": "uv run python ../scripts/gui/run_live_apphost_acceptance.py --live-provider --evidence-dir gui/test-results/live-provider-desktop", "preview": "vite preview", "tauri": "tauri" }, diff --git a/gui/src-tauri/src/app_client/attachment.rs b/gui/src-tauri/src/app_client/attachment.rs index 974a10382..ea3ec15b8 100644 --- a/gui/src-tauri/src/app_client/attachment.rs +++ b/gui/src-tauri/src/app_client/attachment.rs @@ -2,7 +2,7 @@ use super::transport_auth::Channel; use serde::{Deserialize, Serialize}; use serde_json::{json, value::RawValue, Value}; -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use std::io::{Read, Write}; #[path = "projection.rs"] mod projection; @@ -221,6 +221,25 @@ pub(crate) struct ReadSession { closed: bool, } +#[derive(Clone)] +pub(crate) struct ControlAuthority { + attachment_id: String, + controller_generation: String, + instance: String, + members: BTreeMap, +} + +pub(crate) enum ControlResult { + Accepted, + Missing, + Rejected(String), +} + +pub(crate) enum ModelControlResult { + Accepted(Value), + Rejected(String), +} + #[derive(Clone, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct InitialSnapshot { @@ -309,6 +328,36 @@ impl ReadSession { result } + pub(crate) fn control_authority(&self) -> Result { + if self.closed { + return Err(()); + } + let mut members = BTreeMap::new(); + for session in &self.attachment.sessions { + let identity = projection::identity_value(session.member.session.get())?; + let session_id = identity["sessionId"].as_str().ok_or(())?.to_owned(); + if members + .insert(session_id, session.member.member_id.clone()) + .is_some() + { + return Err(()); + } + } + Ok(ControlAuthority { + attachment_id: self.attachment.attachment_id.clone(), + controller_generation: self.attachment.controller_generation.get().to_owned(), + instance: self.instance.clone(), + members, + }) + } + + pub(crate) fn next_control_request_id(&mut self) -> Result { + if self.closed { + return Err(()); + } + take_request_id(&mut self.next_id) + } + /// One complete round, with no hard-coded lifetime or round count. /// No partial per-member cursor changes survive a failed round. pub(crate) fn poll( @@ -475,6 +524,222 @@ impl ReadSession { } } +impl ControlAuthority { + fn app_payload(&self, session_id: &str) -> Result { + let member_id = self.members.get(session_id).ok_or(())?; + let generation: Value = + serde_json::from_str(&self.controller_generation).map_err(|_| ())?; + Ok(json!({ + "attachmentId": self.attachment_id, + "controllerGeneration": generation, + "memberId": member_id, + })) + } + + pub(crate) fn models( + &self, + channel: &mut Channel, + session_id: &str, + request_id: &str, + ) -> Result { + self.model_request(channel, "session/models", session_id, None, request_id) + } + + pub(crate) fn select_model( + &self, + channel: &mut Channel, + session_id: &str, + model_id: &str, + request_id: &str, + ) -> Result { + if model_id.trim().is_empty() || model_id.chars().count() > 512 { + return Err(()); + } + self.model_request( + channel, + "session/model/select", + session_id, + Some(model_id), + request_id, + ) + } + + fn model_request( + &self, + channel: &mut Channel, + operation: &str, + session_id: &str, + model_id: Option<&str>, + request_id: &str, + ) -> Result { + let mut payload = self.app_payload(session_id)?; + if let Some(model_id) = model_id { + payload["modelId"] = Value::String(model_id.to_owned()); + } + let response = request(channel, operation, request_id, payload)?; + decode_model_result(&response.result_type, &response.result) + } + + fn payload(&self, session_id: &str) -> Result { + let member_id = self.members.get(session_id).ok_or(())?; + let generation: Value = + serde_json::from_str(&self.controller_generation).map_err(|_| ())?; + Ok(json!({ + "control": { + "attachmentId": self.attachment_id, + "controllerGeneration": generation, + "memberId": member_id, + }, + "expectedInstanceId": self.instance, + })) + } + + pub(crate) fn submit( + &self, + channel: &mut Channel, + session_id: &str, + submission_id: &str, + text: &str, + request_id: &str, + ) -> Result { + if !identifier(submission_id, 128) + || text.trim().is_empty() + || text.chars().count() > 262_144 + { + return Err(()); + } + let mut payload = self.payload(session_id)?; + payload["submissionId"] = Value::String(submission_id.to_owned()); + payload["text"] = Value::String(text.to_owned()); + let result = execution_request(channel, "execution/submit", request_id, payload)?; + self.record_result(result, session_id, Some(submission_id)) + } + + pub(crate) fn find_submission( + &self, + channel: &mut Channel, + session_id: &str, + submission_id: &str, + request_id: &str, + ) -> Result { + if !identifier(submission_id, 128) { + return Err(()); + } + let mut payload = self.payload(session_id)?; + payload["submissionId"] = Value::String(submission_id.to_owned()); + let result = execution_request(channel, "execution/find_submission", request_id, payload)?; + if result["resultType"] == "not_found" { + projection::control_bridge("not_found", &result.to_string())?; + return Ok(ControlResult::Missing); + } + self.record_result(result, session_id, Some(submission_id)) + } + + pub(crate) fn interrupt( + &self, + channel: &mut Channel, + session_id: &str, + execution_id: &str, + request_id: &str, + ) -> Result { + if !identifier(execution_id, 128) { + return Err(()); + } + let mut payload = self.payload(session_id)?; + payload["executionId"] = Value::String(execution_id.to_owned()); + let result = execution_request(channel, "execution/interrupt", request_id, payload)?; + if let Some(rejected) = rejected_result(&result)? { + return Ok(ControlResult::Rejected(rejected)); + } + let checked = projection::control_bridge("interrupt", &result.to_string())?; + let record = &checked["result"]["record"]; + if record["serviceInstanceId"] != self.instance + || record["identity"]["sessionId"] != session_id + || record["state"]["executionId"] != execution_id + { + return Err(()); + } + Ok(ControlResult::Accepted) + } + + fn record_result( + &self, + result: Value, + session_id: &str, + submission_id: Option<&str>, + ) -> Result { + if let Some(rejected) = rejected_result(&result)? { + return Ok(ControlResult::Rejected(rejected)); + } + let checked = projection::control_bridge("record", &result.to_string())?; + let record = &checked["result"]; + if record["serviceInstanceId"] != self.instance + || record["identity"]["sessionId"] != session_id + || submission_id.is_some_and(|value| record["submissionId"] != value) + { + return Err(()); + } + Ok(ControlResult::Accepted) + } +} + +fn decode_model_result(result_type: &str, result: &RawValue) -> Result { + if result_type == "error" { + let error: Value = serde_json::from_str(result.get()).map_err(|_| ())?; + let code = error.get("code").and_then(Value::as_str).ok_or(())?; + if !identifier(code, 128) { + return Err(()); + } + return Ok(ModelControlResult::Rejected(code.to_owned())); + } + if result_type != "sessionModels" { + return Err(()); + } + let value: Value = serde_json::from_str(result.get()).map_err(|_| ())?; + let root = value.as_object().ok_or(())?; + if root.len() != 2 || !root.contains_key("currentId") || !root.contains_key("models") { + return Err(()); + } + if !root["models"].is_array() + || (!root["currentId"].is_null() && !root["currentId"].is_string()) + { + return Err(()); + } + Ok(ModelControlResult::Accepted(value)) +} + +fn execution_request( + channel: &mut Channel, + operation: &str, + id: &str, + payload: Value, +) -> Result { + let request = serde_json::to_vec(&json!({ + "protocolVersion": "loushang.execution/v1", + "requestId": id, + "operation": operation, + "payload": payload, + })) + .map_err(|_| ())?; + channel.send(&request)?; + let value: Value = serde_json::from_slice(&channel.receive()?).map_err(|_| ())?; + if value["requestId"] != id { + return Err(()); + } + Ok(value) +} + +fn rejected_result(value: &Value) -> Result, ()> { + let kind = value["resultType"].as_str().ok_or(())?; + if kind != "failure" && kind != "app_failure" { + return Ok(None); + } + let checked = projection::control_bridge(kind, &value.to_string())?; + Ok(Some( + checked["result"]["code"].as_str().ok_or(())?.to_owned(), + )) +} + fn successor(decimal: &str) -> String { let mut bytes = decimal.as_bytes().to_vec(); for digit in bytes.iter_mut().rev() { @@ -546,6 +811,16 @@ impl EventReader { #[cfg(test)] mod tests { use super::*; + #[test] + fn model_capability_rejection_is_semantic_not_transport_failure() { + let raw = RawValue::from_string(r#"{"code":"operation_unavailable"}"#.to_owned()).unwrap(); + match decode_model_result("error", &raw).unwrap() { + ModelControlResult::Rejected(code) => assert_eq!(code, "operation_unavailable"), + ModelControlResult::Accepted(_) => panic!("error response was accepted"), + } + assert!(decode_model_result("unexpected", &raw).is_err()); + } + #[test] fn request_numbers_fail_closed_without_wrapping() { let mut next = i64::MAX as u64; diff --git a/gui/src-tauri/src/app_client/mod.rs b/gui/src-tauri/src/app_client/mod.rs index a2f26c16c..af78490d8 100644 --- a/gui/src-tauri/src/app_client/mod.rs +++ b/gui/src-tauri/src/app_client/mod.rs @@ -4,13 +4,14 @@ mod connection_epoch; mod record_native; mod request_deadline; mod transport_auth; +pub(crate) use attachment::{ControlAuthority, ControlResult, ModelControlResult}; use request_deadline::RequestDeadline; use serde::{Deserialize, Serialize}; use std::io::{self, Read, Write}; use std::net::{Ipv4Addr, Shutdown, SocketAddr, TcpStream}; use std::path::Path; -use std::sync::mpsc::{self, Sender}; +use std::sync::mpsc::{self, Receiver, Sender, SyncSender, TryRecvError}; use std::thread::{self, JoinHandle}; use std::time::Duration; @@ -34,6 +35,34 @@ pub(crate) struct LiveEventRound { pub(crate) sessions: Vec, } +pub(crate) enum ControlCommand { + Submit { + session_id: String, + submission_id: String, + text: String, + reply: SyncSender>, + }, + FindSubmission { + session_id: String, + submission_id: String, + reply: SyncSender>, + }, + Interrupt { + session_id: String, + execution_id: String, + reply: SyncSender>, + }, + ListModels { + session_id: String, + reply: SyncSender>, + }, + SelectModel { + session_id: String, + model_id: String, + reply: SyncSender>, + }, +} + struct DeadlineStream { stream: TcpStream, deadline: RequestDeadline, @@ -91,6 +120,7 @@ pub(crate) struct ConnectionOptions<'a> { pub(crate) attach: bool, pub(crate) mux_name: &'a str, pub(crate) stop: Option, + pub(crate) control_commands: Option<&'a Receiver>, } fn run_attempt( @@ -99,6 +129,7 @@ fn run_attempt( connection_epoch: u64, publish_initial: &mut dyn FnMut(LiveInitialSnapshot) -> Result<(), ()>, publish_round: &mut dyn FnMut(LiveEventRound) -> Result<(), ()>, + publish_control: &mut dyn FnMut(Option) -> Result<(), ()>, continue_reading: &mut dyn FnMut() -> Result, ) -> Result { let ConnectionOptions { @@ -108,6 +139,7 @@ fn run_attempt( attach, mux_name, stop, + control_commands, } = options; let deadline = RequestDeadline::new(timeout)?; let attempt = owner.begin()?; @@ -186,7 +218,13 @@ fn run_attempt( attachment::ReadSession::attach(&mut channel, instance, mux_name, &attempt)?; let outcome = (|| { let initial = session.initialize(&mut channel, &attempt)?; + let authority = session.control_authority()?; + let control_authority = authority.clone(); let mux_space_id = initial.mux_space["muxSpaceId"].clone(); + attempt.apply(|| publish_control(Some(authority)))?; + if let Some(commands) = control_commands { + process_controls(&mut session, &control_authority, &mut channel, commands)?; + } attempt.apply(|| { publish_initial(LiveInitialSnapshot { connection_epoch: connection_epoch.to_string(), @@ -199,6 +237,9 @@ fn run_attempt( let mut sequence = 0_u64; if let Some(stop) = stop.as_ref() { while !stop.requested()? && continue_reading()? { + if let Some(commands) = control_commands { + process_controls(&mut session, &control_authority, &mut channel, commands)?; + } let Some(round) = session.poll(&mut channel, &attempt, Some(stop))? else { if stop.requested()? { break; @@ -234,8 +275,10 @@ fn run_attempt( Ok(()) })(); let detached = session.detach(&mut channel, &attempt); + let control_released = publish_control(None); outcome?; detached?; + control_released?; } stream.shutdown(Shutdown::Both).map_err(|_| ())?; attempt.apply(|| Ok(hello.service_instance_id)) @@ -257,6 +300,7 @@ impl AppClient { 1, publish, &mut |_| unreachable!(), + &mut |_| Ok(()), &mut || Ok(true), ) } @@ -268,6 +312,7 @@ impl AppClient { connection_epoch: u64, publish_initial: &mut dyn FnMut(LiveInitialSnapshot) -> Result<(), ()>, publish_round: &mut dyn FnMut(LiveEventRound) -> Result<(), ()>, + publish_control: &mut dyn FnMut(Option) -> Result<(), ()>, continue_reading: &mut dyn FnMut() -> Result, ) -> Result { run_attempt( @@ -276,11 +321,82 @@ impl AppClient { connection_epoch, publish_initial, publish_round, + publish_control, continue_reading, ) } } +fn process_controls( + session: &mut attachment::ReadSession, + authority: &ControlAuthority, + channel: &mut transport_auth::Channel, + commands: &Receiver, +) -> Result<(), ()> { + loop { + let command = match commands.try_recv() { + Ok(command) => command, + Err(TryRecvError::Empty) => return Ok(()), + Err(TryRecvError::Disconnected) => return Err(()), + }; + let request_id = session.next_control_request_id()?; + let failed = match command { + ControlCommand::Submit { + session_id, + submission_id, + text, + reply, + } => { + let result = + authority.submit(channel, &session_id, &submission_id, &text, &request_id); + let failed = result.is_err(); + let _ = reply.send(result); + failed + } + ControlCommand::FindSubmission { + session_id, + submission_id, + reply, + } => { + let result = + authority.find_submission(channel, &session_id, &submission_id, &request_id); + let failed = result.is_err(); + let _ = reply.send(result); + failed + } + ControlCommand::Interrupt { + session_id, + execution_id, + reply, + } => { + let result = authority.interrupt(channel, &session_id, &execution_id, &request_id); + let failed = result.is_err(); + let _ = reply.send(result); + failed + } + ControlCommand::ListModels { session_id, reply } => { + let result = authority.models(channel, &session_id, &request_id); + let failed = result.is_err(); + let _ = reply.send(result); + failed + } + ControlCommand::SelectModel { + session_id, + model_id, + reply, + } => { + let result = authority.select_model(channel, &session_id, &model_id, &request_id); + let failed = result.is_err(); + let _ = reply.send(result); + failed + } + }; + if failed { + return Err(()); + } + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/gui/src-tauri/src/app_client/projection.rs b/gui/src-tauri/src/app_client/projection.rs index 3ed2fa5b1..52c6bb32c 100644 --- a/gui/src-tauri/src/app_client/projection.rs +++ b/gui/src-tauri/src/app_client/projection.rs @@ -221,6 +221,25 @@ struct Snapshot { } #[derive(Deserialize, Serialize)] #[serde(rename_all = "camelCase", deny_unknown_fields)] +struct RecordEnvelope { + service_instance_id: String, + identity: Identity, + submission_id: Box, + state: State, +} +#[derive(Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct InterruptResult { + record: RecordEnvelope, + disposition: String, +} +#[derive(Deserialize, Serialize)] +#[serde(deny_unknown_fields)] +struct Failure { + code: String, +} +#[derive(Deserialize, Serialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] struct Event { session_id: String, cursor: Counter, @@ -335,6 +354,86 @@ pub fn bridge(kind: &str, wire: &str) -> Result { } } +pub fn control_bridge(kind: &str, wire: &str) -> Result { + if kind == "record" { + let parsed: Envelope = serde_json::from_str(wire).map_err(|_| ())?; + let submission = optional_string(&parsed.result.submission_id)?; + if !envelope(&parsed, "record") + || !identifier(&parsed.result.service_instance_id, 128) + || !parsed.result.identity.valid() + || submission + .as_ref() + .is_some_and(|value| !identifier(value, 128)) + || !parsed.result.state.valid() + { + return Err(()); + } + return serde_json::to_value(parsed).map_err(|_| ()); + } + if kind == "interrupt" { + let parsed: Envelope = serde_json::from_str(wire).map_err(|_| ())?; + let record = &parsed.result.record; + let submission = optional_string(&record.submission_id)?; + if !envelope(&parsed, "interrupt") + || !identifier(&record.service_instance_id, 128) + || !record.identity.valid() + || submission + .as_ref() + .is_some_and(|item| !identifier(item, 128)) + || !record.state.valid() + || !["requested", "already_terminal"].contains(&parsed.result.disposition.as_str()) + || (parsed.result.disposition == "requested" && !record.state.interrupt_requested) + || (parsed.result.disposition == "already_terminal" && !terminal(&record.state.status)) + { + return Err(()); + } + return serde_json::to_value(parsed).map_err(|_| ()); + } + if kind == "not_found" { + let parsed: Envelope> = serde_json::from_str(wire).map_err(|_| ())?; + if !envelope(&parsed, "not_found") || parsed.result.get() != "null" { + return Err(()); + } + return serde_json::to_value(parsed).map_err(|_| ()); + } + if kind == "failure" || kind == "app_failure" { + let parsed: Envelope = serde_json::from_str(wire).map_err(|_| ())?; + let allowed = if kind == "failure" { + [ + "service_instance_changed", + "submission_conflict", + "submission_ledger_full", + "execution_busy", + "execution_not_retained", + "execution_unsupported", + ] + .contains(&parsed.result.code.as_str()) + } else { + [ + "invalid_request", + "not_found", + "already_exists", + "already_attached", + "product_mismatch", + "revision_conflict", + "snapshot_required", + "stale_attachment", + "attachment_lagged", + "session_unavailable", + "operation_unavailable", + "cleanup_incomplete", + "service_closed", + ] + .contains(&parsed.result.code.as_str()) + }; + if !envelope(&parsed, kind) || !allowed { + return Err(()); + } + return serde_json::to_value(parsed).map_err(|_| ()); + } + Err(()) +} + #[allow(dead_code)] // Used by the attachment binary, not the value-only binary. pub fn source_value(wire: &str) -> Result { let source: Source = serde_json::from_str(wire).map_err(|_| ())?; diff --git a/gui/src-tauri/src/lib.rs b/gui/src-tauri/src/lib.rs index 0e62ff4d5..47482b7aa 100644 --- a/gui/src-tauri/src/lib.rs +++ b/gui/src-tauri/src/lib.rs @@ -50,6 +50,10 @@ pub fn run() { let builder = builder.invoke_handler(tauri::generate_handler![ fixture_bridge_handshake, live_bridge::live_readonly_available, + live_bridge::live_submit_text, + live_bridge::live_interrupt, + live_bridge::live_session_models, + live_bridge::live_select_model, live_bridge::resync_live_readonly, live_bridge::start_live_readonly, live_bridge::stop_live_readonly @@ -57,6 +61,10 @@ pub fn run() { #[cfg(all(windows, not(feature = "fixture-bridge")))] let builder = builder.invoke_handler(tauri::generate_handler![ live_bridge::live_readonly_available, + live_bridge::live_submit_text, + live_bridge::live_interrupt, + live_bridge::live_session_models, + live_bridge::live_select_model, live_bridge::resync_live_readonly, live_bridge::start_live_readonly, live_bridge::stop_live_readonly diff --git a/gui/src-tauri/src/live_bridge.rs b/gui/src-tauri/src/live_bridge.rs index f31c145ca..a915ae89a 100644 --- a/gui/src-tauri/src/live_bridge.rs +++ b/gui/src-tauri/src/live_bridge.rs @@ -1,11 +1,14 @@ -//! Tauri handoff for reconnecting read-only snapshots and validated event rounds. -use crate::app_client::{AppClient, ConnectionOptions, LiveEventRound, LiveInitialSnapshot}; +//! Tauri handoff for reconnecting snapshots, validated event rounds and bounded control. +use crate::app_client::{ + AppClient, ConnectionOptions, ControlCommand, ControlResult, LiveEventRound, + LiveInitialSnapshot, ModelControlResult, +}; use crate::connection_lifecycle::{ConnectionLifecycle, ExitAction}; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use std::ffi::OsString; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; -use std::sync::{mpsc, Arc}; +use std::sync::{mpsc, Arc, RwLock}; use std::time::Duration; use tauri::{Emitter, Manager}; @@ -23,6 +26,13 @@ struct LiveLaunch { pub(crate) struct LiveLaunchState { launch: Result, ()>, resync: Arc, + control: Arc>>, +} + +#[derive(Clone)] +struct LiveControl { + connection_epoch: u64, + sender: mpsc::Sender, } impl LiveLaunchState { @@ -30,6 +40,7 @@ impl LiveLaunchState { Self { launch: parse_launch(std::env::args_os().skip(1)), resync: Arc::new(AtomicU64::new(0)), + control: Arc::new(RwLock::new(None)), } } } @@ -41,6 +52,41 @@ struct ConnectionState { connection_epoch: String, } +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct LiveSubmitInput { + session_id: String, + submission_id: String, + text: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct LiveInterruptInput { + session_id: String, + execution_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct LiveModelsInput { + session_id: String, +} + +#[derive(Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub(crate) struct LiveSelectModelInput { + session_id: String, + model_id: String, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct LiveControlReceipt { + accepted: bool, + reason: Option, +} + enum Publication { Initial(LiveInitialSnapshot), Round(LiveEventRound), @@ -92,6 +138,7 @@ pub(crate) fn start_live_readonly(app: tauri::AppHandle) -> Result<(), String> { .clone() .ok_or_else(|| "live AppHost launch configuration is unavailable".to_owned())?; let resync = state.resync.clone(); + let control = state.control.clone(); let owner = app.state::>().inner().clone(); let (publish, receive) = mpsc::channel::(); owner @@ -102,6 +149,8 @@ pub(crate) fn start_live_readonly(app: tauri::AppHandle) -> Result<(), String> { while !stop.requested()? { epoch = epoch.checked_add(1).ok_or(())?; let token = resync.load(Ordering::Acquire); + let (control_sender, control_receiver) = mpsc::channel(); + *control.write().map_err(|_| ())? = None; publish .send(Publication::State(ConnectionState { state: "connecting", @@ -117,6 +166,7 @@ pub(crate) fn start_live_readonly(app: tauri::AppHandle) -> Result<(), String> { attach: true, mux_name: &launch.mux_name, stop: Some(stop.clone()), + control_commands: Some(&control_receiver), }, epoch, &mut |snapshot| { @@ -124,8 +174,16 @@ pub(crate) fn start_live_readonly(app: tauri::AppHandle) -> Result<(), String> { publish.send(Publication::Initial(snapshot)).map_err(|_| ()) }, &mut |round| publish.send(Publication::Round(round)).map_err(|_| ()), + &mut |authority| { + *control.write().map_err(|_| ())? = authority.map(|_| LiveControl { + connection_epoch: epoch, + sender: control_sender.clone(), + }); + Ok(()) + }, &mut || Ok(resync.load(Ordering::Acquire) == token), ); + *control.write().map_err(|_| ())? = None; if stop.requested()? { break; } @@ -164,6 +222,192 @@ pub(crate) fn start_live_readonly(app: tauri::AppHandle) -> Result<(), String> { Ok(()) } +#[tauri::command] +pub(crate) async fn live_submit_text( + state: tauri::State<'_, LiveLaunchState>, + input: LiveSubmitInput, +) -> Result { + let control = current_control(&state)?; + let controls = state.control.clone(); + tauri::async_runtime::spawn_blocking(move || { + let recovery_session_id = input.session_id.clone(); + let recovery_submission_id = input.submission_id.clone(); + let (reply, receive) = mpsc::sync_channel(1); + control + .sender + .send(ControlCommand::Submit { + session_id: input.session_id, + submission_id: input.submission_id, + text: input.text, + reply, + }) + .map_err(|_| "live control connection changed".to_owned())?; + let result = receive + .recv_timeout(Duration::from_secs(6)) + .map_err(|_| "live submit response timed out".to_owned())?; + match result { + Ok(result) => control_receipt(Ok(result), true), + Err(()) => { + let recovered = wait_for_reconnected_control(&controls, control.connection_epoch)?; + let (reply, receive) = mpsc::sync_channel(1); + recovered + .sender + .send(ControlCommand::FindSubmission { + session_id: recovery_session_id, + submission_id: recovery_submission_id, + reply, + }) + .map_err(|_| "live recovery connection changed".to_owned())?; + let result = receive + .recv_timeout(Duration::from_secs(6)) + .map_err(|_| "live submission recovery timed out".to_owned())?; + control_receipt(result, true) + } + } + }) + .await + .map_err(|_| "live submit worker failed".to_owned())? +} + +#[tauri::command] +pub(crate) async fn live_interrupt( + state: tauri::State<'_, LiveLaunchState>, + input: LiveInterruptInput, +) -> Result { + let control = current_control(&state)?; + tauri::async_runtime::spawn_blocking(move || { + let (reply, receive) = mpsc::sync_channel(1); + control + .sender + .send(ControlCommand::Interrupt { + session_id: input.session_id, + execution_id: input.execution_id, + reply, + }) + .map_err(|_| "live control connection changed".to_owned())?; + let result = receive + .recv_timeout(Duration::from_secs(6)) + .map_err(|_| "live interrupt response timed out".to_owned())?; + control_receipt(result, false) + }) + .await + .map_err(|_| "live interrupt worker failed".to_owned())? +} + +#[tauri::command] +pub(crate) async fn live_session_models( + state: tauri::State<'_, LiveLaunchState>, + input: LiveModelsInput, +) -> Result { + let control = current_control(&state)?; + tauri::async_runtime::spawn_blocking(move || { + let (reply, receive) = mpsc::sync_channel(1); + control + .sender + .send(ControlCommand::ListModels { + session_id: input.session_id, + reply, + }) + .map_err(|_| "live control connection changed".to_owned())?; + match receive + .recv_timeout(Duration::from_secs(6)) + .map_err(|_| "model listing timed out".to_owned())? + .map_err(|_| "model listing connection failed".to_owned())? + { + ModelControlResult::Accepted(value) => Ok(value), + ModelControlResult::Rejected(code) => Err(format!("model listing rejected: {code}")), + } + }) + .await + .map_err(|_| "model listing worker failed".to_owned())? +} + +#[tauri::command] +pub(crate) async fn live_select_model( + state: tauri::State<'_, LiveLaunchState>, + input: LiveSelectModelInput, +) -> Result { + let control = current_control(&state)?; + tauri::async_runtime::spawn_blocking(move || { + let (reply, receive) = mpsc::sync_channel(1); + control + .sender + .send(ControlCommand::SelectModel { + session_id: input.session_id, + model_id: input.model_id, + reply, + }) + .map_err(|_| "live control connection changed".to_owned())?; + match receive + .recv_timeout(Duration::from_secs(6)) + .map_err(|_| "model selection timed out".to_owned())? + .map_err(|_| "model selection connection failed".to_owned())? + { + ModelControlResult::Accepted(value) => Ok(value), + ModelControlResult::Rejected(code) => Err(format!("model selection rejected: {code}")), + } + }) + .await + .map_err(|_| "model selection worker failed".to_owned())? +} + +fn current_control(state: &LiveLaunchState) -> Result { + let control = state + .control + .read() + .map_err(|_| "live control state is unavailable".to_owned())? + .clone() + .ok_or_else(|| "live control is not attached".to_owned())?; + let _epoch = control.connection_epoch; + Ok(control) +} + +fn wait_for_reconnected_control( + controls: &Arc>>, + previous_epoch: u64, +) -> Result { + let deadline = std::time::Instant::now() + .checked_add(Duration::from_secs(6)) + .ok_or_else(|| "live recovery deadline is unavailable".to_owned())?; + loop { + let current = controls + .read() + .map_err(|_| "live control state is unavailable".to_owned())? + .clone(); + if let Some(control) = current { + if control.connection_epoch > previous_epoch { + return Ok(control); + } + } + if std::time::Instant::now() >= deadline { + return Err("live control did not reconnect for submission recovery".to_owned()); + } + std::thread::sleep(Duration::from_millis(50)); + } +} + +fn control_receipt( + result: Result, + submit: bool, +) -> Result { + match result { + Ok(ControlResult::Accepted) => Ok(LiveControlReceipt { + accepted: true, + reason: None, + }), + Ok(ControlResult::Rejected(reason)) => Ok(LiveControlReceipt { + accepted: false, + reason: Some(reason), + }), + Ok(ControlResult::Missing) if submit => Ok(LiveControlReceipt { + accepted: false, + reason: Some("submission_outcome_unknown".to_owned()), + }), + Ok(ControlResult::Missing) => Err("interrupt result is missing".to_owned()), + Err(()) => Err("live control request failed".to_owned()), + } +} + #[tauri::command] pub(crate) fn resync_live_readonly(app: tauri::AppHandle) -> Result<(), String> { app.state::() @@ -177,6 +421,9 @@ pub(crate) fn resync_live_readonly(app: tauri::AppHandle) -> Result<(), String> #[tauri::command] pub(crate) fn stop_live_readonly(app: tauri::AppHandle) { + if let Ok(mut control) = app.state::().control.write() { + *control = None; + } let owner = app.state::>().inner().clone(); if let ExitAction::Join(worker) = owner.begin_exit() { tauri::async_runtime::spawn_blocking(move || owner.finish_exit(worker.wait())); diff --git a/gui/src/App.tsx b/gui/src/App.tsx index 749397974..76af53cf4 100644 --- a/gui/src/App.tsx +++ b/gui/src/App.tsx @@ -14,6 +14,7 @@ import type { ReadonlyDocument, RunProjection, SessionSnapshot, + SessionModels, TaskProjection, WorkspaceSummary, } from "./client/model"; @@ -65,6 +66,14 @@ export function HarnessGui({ client }: HarnessGuiProps) { const refreshRef = useRef<(() => void) | null>(null); const syncPendingRef = useRef(true); const [remainingSteps, setRemainingSteps] = useState(fixture?.remainingFixtureSteps() ?? 0); + const [controlPending, setControlPending] = useState<"submit" | "interrupt" | null>(null); + const [sessionModels, setSessionModels] = useState(null); + const [modelPending, setModelPending] = useState(false); + const controlPendingRef = useRef(false); + const [awaitingPublication, setAwaitingPublication] = useState<{ + readonly sessionId: string; + readonly cursor: string; + } | null>(null); const [bridgeProbe, setBridgeProbe] = useState({ status: "web-mock", label: "Web Mock", @@ -97,7 +106,7 @@ export function HarnessGui({ client }: HarnessGuiProps) { setRemainingSteps(fixture?.remainingFixtureSteps() ?? 0); setNotice(snapshot.source.kind === "fixture" ? "Fixture snapshot installed. No backend is running." - : "Read-only live snapshot installed."); + : "Live AppHost connection installed."); } catch { if (active) dispatch({ type: "sync.failed", @@ -126,7 +135,7 @@ export function HarnessGui({ client }: HarnessGuiProps) { const unsubscribeSnapshots = client.subscribeSnapshots?.((snapshot) => { if (!active) return; dispatch({ type: "snapshot.installed", snapshot }); - setNotice("Read-only live event round installed."); + setNotice("Live AppHost event round installed."); setSyncing(false); }) ?? (() => undefined); const unsubscribeConnection = client.subscribeConnection?.((connection) => { @@ -186,8 +195,29 @@ export function HarnessGui({ client }: HarnessGuiProps) { const stackedRight = rightVisible && remainingWidth < rightWidth + 360; const tasksAvailable = hasCapability(state, "tasks"); const changesAvailable = hasCapability(state, "changes"); - const mutationsAllowed = !syncing && state.remote.source.kind === "fixture" && state.remote.connection === "fixture-offline"; const selected = selectedId ? state.remote.sessions[selectedId] : undefined; + useEffect(() => { + let active = true; + setSessionModels(null); + if (!selectedId || state.remote.connection !== "connected" || !client.sessionModels) return () => { active = false; }; + void client.sessionModels(selectedId).then((models) => { + if (active) setSessionModels(models); + }).catch(() => { + if (active) setNotice("The AppHost model catalog is unavailable."); + }); + return () => { active = false; }; + }, [client, selectedId, state.remote.connection]); + useEffect(() => { + if (!awaitingPublication) return; + const session = state.remote.sessions[awaitingPublication.sessionId]; + if (state.remote.connection !== "connected" || !session || session.cursor !== awaitingPublication.cursor) { + setAwaitingPublication(null); + } + }, [awaitingPublication, state.remote.connection, state.remote.sessions]); + const mutationsAllowed = !syncing && !controlPending && !awaitingPublication && ( + (state.remote.source.kind === "fixture" && state.remote.connection === "fixture-offline") + || (state.remote.source.kind === "live" && state.remote.connection === "connected") + ); useLayoutEffect(() => { const pane = conversationRef.current; const transcript = transcriptRef.current; @@ -214,21 +244,37 @@ export function HarnessGui({ client }: HarnessGuiProps) { ); async function submit(): Promise { - if (!selected || syncPendingRef.current || !mutationsAllowed || selected.status === "running") return; + if (!selected || syncPendingRef.current || controlPendingRef.current || !mutationsAllowed || selected.status === "running") return; const text = state.local.drafts[selected.id] ?? ""; if (!text.trim()) { setNotice("Enter a fixture prompt before sending."); return; } - const submissionId = `gui-fixture-submission-${submissionCounter.current++}`; - const receipt = await client.submitText({ sessionId: selected.id, submissionId, text }); - if (!receipt.accepted) { - setNotice("The fixture rejected this submission."); - return; + const submissionId = fixture + ? `gui-fixture-submission-${submissionCounter.current++}` + : `gui-${crypto.randomUUID()}`; + controlPendingRef.current = true; + setControlPending("submit"); + try { + const receipt = await client.submitText({ sessionId: selected.id, submissionId, text }); + if (!receipt.accepted) { + setNotice(receipt.reason === "submission_outcome_unknown" + ? "Submission outcome is unknown. The draft was preserved; resynchronize before retrying." + : `Submission rejected${receipt.reason ? `: ${receipt.reason}` : "."}`); + return; + } + dispatch({ type: "draft.submitted", sessionId: selected.id, expected: text }); + if (!fixture) setAwaitingPublication({ sessionId: selected.id, cursor: selected.cursor }); + setRemainingSteps(fixture?.remainingFixtureSteps() ?? 0); + setNotice(fixture + ? "Fixture accepted the prompt. Advance it step by step." + : "AppHost accepted the prompt."); + } catch { + setNotice("Submit failed before its outcome could be confirmed. The draft was preserved."); + } finally { + controlPendingRef.current = false; + setControlPending(null); } - dispatch({ type: "draft.changed", sessionId: selected.id, value: "" }); - setRemainingSteps(fixture?.remainingFixtureSteps() ?? 0); - setNotice("Fixture accepted the prompt. Advance it step by step."); } async function advance(): Promise { @@ -243,14 +289,36 @@ export function HarnessGui({ client }: HarnessGuiProps) { } async function interrupt(): Promise { - if (!selected || syncPendingRef.current || !mutationsAllowed || selected.status !== "running") return; - const receipt = await client.interrupt(selected.id); - setRemainingSteps(fixture?.remainingFixtureSteps() ?? 0); - setNotice( - receipt.accepted - ? "Fixture execution interrupted." - : (receipt.reason ?? "Interrupt was not accepted."), - ); + if (!selected || syncPendingRef.current || controlPendingRef.current || !mutationsAllowed || selected.status !== "running") return; + controlPendingRef.current = true; + setControlPending("interrupt"); + try { + const receipt = await client.interrupt(selected.id); + setRemainingSteps(fixture?.remainingFixtureSteps() ?? 0); + setNotice(receipt.accepted + ? (fixture ? "Fixture execution interrupted." : "Interrupt requested.") + : (receipt.reason ?? "Interrupt was not accepted.")); + } catch { + setNotice("Interrupt failed. Resynchronize before retrying."); + } finally { + controlPendingRef.current = false; + setControlPending(null); + } + } + + async function selectModel(modelId: string): Promise { + if (!selected || !client.selectModel || selected.status === "running" || modelPending) return; + setModelPending(true); + try { + const models = await client.selectModel(selected.id, modelId); + setSessionModels(models); + const choice = models.models.find((model) => model.id === models.currentId); + setNotice(`Session model changed to ${choice?.label ?? modelId}.`); + } catch { + setNotice("Model change was rejected. The Session kept its previous model."); + } finally { + setModelPending(false); + } } function openReview(documentId: string): void { @@ -277,7 +345,7 @@ export function HarnessGui({ client }: HarnessGuiProps) { bridgeProbe={bridgeProbe} onSelect={(sessionId) => dispatch({ type: "session.selected", sessionId })} onToggleWorkspace={(workspaceId) => dispatch({ type: "workspace.toggled", workspaceId })} - onUnavailable={() => setNotice("New Session is unavailable in this read-only slice.")} + onUnavailable={() => setNotice("New Session is unavailable in this existing-Session slice.")} />
{syncing || state.diagnostic || state.remote.connection === "disconnected" ? (
- {syncing ? (fixture ? "Synchronizing fixture…" : "Connecting read-only…") : state.diagnostic ?? "Disconnected."} + {syncing ? (fixture ? "Synchronizing fixture…" : "Connecting to AppHost…") : state.diagnostic ?? "Disconnected."} {fixture ? : null}
) : null} @@ -384,8 +452,7 @@ export function HarnessGui({ client }: HarnessGuiProps) { id="message-input" rows={2} value={state.local.drafts[selected.id] ?? ""} - placeholder={state.remote.source.kind === "fixture" ? "Type a fixture prompt…" : "Read-only live connection"} - disabled={state.remote.source.kind === "live"} + placeholder={state.remote.source.kind === "fixture" ? "Type a fixture prompt…" : "Message the live Agent…"} onChange={(event) => dispatch({ type: "draft.changed", sessionId: selected.id, value: event.currentTarget.value }) } @@ -398,9 +465,23 @@ export function HarnessGui({ client }: HarnessGuiProps) { />
- ◉ {state.remote.source.kind === "fixture" ? "Fixture access" : "Read-only"} + ◉ {state.remote.source.kind === "fixture" ? "Fixture access" : "AppHost control"}

{notice}

- + -
{state.remote.source.kind === "fixture" ? "Fixture data · no backend" : "Live AppHost · read-only"}
+
{state.remote.source.kind === "fixture" ? "Fixture data · no backend" : "Live AppHost · controlled"}