diff --git a/crates/codeoid-client/src/connection.rs b/crates/codeoid-client/src/connection.rs index c1a2737..54f785b 100644 --- a/crates/codeoid-client/src/connection.rs +++ b/crates/codeoid-client/src/connection.rs @@ -525,6 +525,8 @@ fn daemon_kind(msg: &DaemonMessage) -> &'static str { DaemonMessage::SettingsSchemaResult { .. } => "settings.schema.result", DaemonMessage::SettingsGetResult { .. } => "settings.get.result", DaemonMessage::SettingsSetResult { .. } => "settings.set.result", + DaemonMessage::FleetSnapshotResult { .. } => "fleet.snapshot.result", + DaemonMessage::FleetUpdate { .. } => "fleet.update", DaemonMessage::Unknown => "unknown", } } @@ -569,6 +571,8 @@ fn client_kind(msg: &ClientMessage) -> &'static str { ClientMessage::SettingsSchema { .. } => "settings.schema", ClientMessage::SettingsGet { .. } => "settings.get", ClientMessage::SettingsSet { .. } => "settings.set", + ClientMessage::FleetSubscribe { .. } => "fleet.subscribe", + ClientMessage::FleetUnsubscribe { .. } => "fleet.unsubscribe", } } diff --git a/crates/codeoid-protocol/src/client.rs b/crates/codeoid-protocol/src/client.rs index c06b1bd..69326cb 100644 --- a/crates/codeoid-protocol/src/client.rs +++ b/crates/codeoid-protocol/src/client.rs @@ -274,6 +274,28 @@ pub enum ClientMessage { id: String, patches: Vec, }, + + /// Subscribe to the fleet board: answered with `fleet.snapshot.result`, + /// then streamed `fleet.update` deltas until `fleet.unsubscribe` or the + /// socket drops. Gated on the `fleet:read` scope. + #[serde(rename = "fleet.subscribe", rename_all = "camelCase")] + FleetSubscribe { + id: String, + /// Only `Tenant` today — the caller's own account+project board. + scope: FleetScope, + }, + + /// Stop the delta stream without dropping the connection. + #[serde(rename = "fleet.unsubscribe", rename_all = "camelCase")] + FleetUnsubscribe { id: String }, +} + +/// Breadth of a fleet subscription. Closed on purpose: widening it is an +/// explicit protocol change on both sides, not something a client can ask for. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum FleetScope { + Tenant, } /// One change requested by `settings.set`, addressed by field `key`. The @@ -326,7 +348,9 @@ impl ClientMessage { | Self::SessionImport { id, .. } | Self::SettingsSchema { id } | Self::SettingsGet { id } - | Self::SettingsSet { id, .. } => id, + | Self::SettingsSet { id, .. } + | Self::FleetSubscribe { id, .. } + | Self::FleetUnsubscribe { id } => id, } } } diff --git a/crates/codeoid-protocol/src/daemon.rs b/crates/codeoid-protocol/src/daemon.rs index c7bf0ac..2f90190 100644 --- a/crates/codeoid-protocol/src/daemon.rs +++ b/crates/codeoid-protocol/src/daemon.rs @@ -200,11 +200,152 @@ pub enum DaemonMessage { restart_required: bool, }, + /// Reply to `fleet.subscribe` — the whole board in one payload. + #[serde(rename = "fleet.snapshot.result", rename_all = "camelCase")] + FleetSnapshotResult { + request_id: String, + fleet: FleetSnapshot, + }, + + /// One incremental board change, pushed to subscribed clients. + #[serde(rename = "fleet.update", rename_all = "camelCase")] + FleetUpdate { delta: FleetDelta }, + /// Forward-compat sink. Preserves raw JSON so the TUI can log it. #[serde(other)] Unknown, } +// ── Fleet board (mirrors codeoid/packages/protocol types.ts) ───────────────── + +/// A dispatch task as the board draws it. +/// +/// Note what is NOT here: the dispatch `prompt` and the worker `workdir`. The +/// daemon deliberately withholds them — the board renders lifecycle, and the +/// prompt is the one field on a task row carrying arbitrary user text. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct FleetTask { + pub id: String, + pub kind: FleetTaskKind, + pub shape: FleetTaskShape, + pub status: FleetTaskStatus, + pub attempts: u32, + /// Epoch ms. + pub created_at: i64, + /// spawn: the worker session this task created. Joins to `FleetSnapshot::workers`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub worker_session_id: Option, + /// send: the existing session this task was routed to. Same join. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub target_session: Option, + /// Compressed result — never a raw transcript. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub result_digest: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, + /// Conductor WIMSE URI — who dispatched this. + pub created_by: String, + /// Dispatch group (fan-out barrier); absent = standalone. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub group_id: Option, + /// RESERVED — never populated by the daemon today. Present so typed + /// fan-in edges are a later non-breaking add. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub depends_on: Option>, +} + +/// `#[serde(other)]` throughout: a daemon newer than this client must degrade +/// to an unrendered node, never fail the whole board's deserialization. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum FleetTaskKind { + Send, + Spawn, + #[serde(other)] + Unknown, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum FleetTaskShape { + /// Deliver a change. + Ship, + /// Investigate and report; never pushes. + Scout, + #[serde(other)] + Unknown, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum FleetTaskStatus { + Queued, + Claimed, + Running, + Done, + Failed, + /// The failure cap tripped — needs a human, will not retry itself. + Blocked, + #[serde(other)] + Unknown, +} + +/// A dispatch lifecycle event — the audit trail behind the board. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct FleetEvent { + pub id: i64, + pub task_id: String, + #[serde(rename = "type")] + pub event_type: String, + pub digest: String, + /// Epoch ms. + pub created_at: i64, +} + +/// Fleet-wide rollup, normalized across backends. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct FleetUsage { + pub active_tasks: u32, + pub blocked_tasks: u32, + pub input_tokens: u64, + pub output_tokens: u64, + pub total_cost_usd: f64, +} + +/// Everything needed to draw the fleet, in one payload. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct FleetSnapshot { + /// Absent when the tenant has no conductor — a valid, common state. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub conductor: Option, + /// Sessions the board references — spawned workers and dispatch targets. + #[serde(default)] + pub workers: Vec, + /// Newest first. + #[serde(default)] + pub tasks: Vec, + /// Newest first. + #[serde(default)] + pub events: Vec, + #[serde(default)] + pub agg: FleetUsage, +} + +/// One incremental board change. Carries the FULL row rather than a patch, so a +/// client that missed a delta still converges and re-delivery is idempotent. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "lowercase")] +pub enum FleetDelta { + #[serde(rename_all = "camelCase")] + Task { task: FleetTask, agg: FleetUsage }, + #[serde(rename_all = "camelCase")] + Event { event: FleetEvent, agg: FleetUsage }, +} + // ── Settings manifest + snapshot (mirrors codeoid/packages/protocol settings.ts) ── /// The declarative settings manifest served over `settings.schema`. Rendered diff --git a/crates/codeoid-protocol/src/lib.rs b/crates/codeoid-protocol/src/lib.rs index aafdd83..874005a 100644 --- a/crates/codeoid-protocol/src/lib.rs +++ b/crates/codeoid-protocol/src/lib.rs @@ -38,22 +38,25 @@ pub mod session; pub mod tool; pub use client::{ - Attachment, ClientMessage, SearchScope, SendPriority, SessionImportSource, SettingPatch, + Attachment, ClientMessage, FleetScope, SearchScope, SendPriority, SessionImportSource, + SettingPatch, }; pub use daemon::{ AuthOkMsg, ClaudeConfigAgent, ClaudeConfigHook, ClaudeConfigMcpServer, ClaudeConfigScope, - ClaudeConfigSkill, DaemonMessage, ErrorCode, McpServerStatus, ModelInfo, ProviderCommand, - SecretStatus, SessionExportCounts, SessionExportManifest, SessionExportMetaSlim, - SessionExportPayload, SessionExportWorkdir, SessionSearchHit, SessionSearchSnippet, - SessionUiRequestMsg, SettingError, SettingField, SettingOption, SettingState, SettingsGroup, - SettingsManifest, SettingsSnapshot, SettingsTab, UiRequestMethod, UiResolvedReason, + ClaudeConfigSkill, DaemonMessage, ErrorCode, FleetDelta, FleetEvent, FleetSnapshot, FleetTask, + FleetTaskKind, FleetTaskShape, FleetTaskStatus, FleetUsage, McpServerStatus, ModelInfo, + ProviderCommand, SecretStatus, SessionExportCounts, SessionExportManifest, + SessionExportMetaSlim, SessionExportPayload, SessionExportWorkdir, SessionSearchHit, + SessionSearchSnippet, SessionUiRequestMsg, SettingError, SettingField, SettingOption, + SettingState, SettingsGroup, SettingsManifest, SettingsSnapshot, SettingsTab, UiRequestMethod, + UiResolvedReason, }; pub use message::{ ContentPart, IdentityType, MessageIdentity, MessageRole, SessionMessage, SessionMessageDelta, }; pub use session::{ CollaborationConfig, CollaborationRole, CollaborationRoleRef, ForkedFrom, SessionInfo, - SessionMode, SessionStatus, SessionUsage, SessionWorktree, Subagent, TurnUsage, + SessionMode, SessionRole, SessionStatus, SessionUsage, SessionWorktree, Subagent, TurnUsage, }; pub use tool::{CancelReason, ConfirmedBy, ToolInfo, ToolPhase, ToolState}; diff --git a/crates/codeoid-protocol/src/session.rs b/crates/codeoid-protocol/src/session.rs index 1384456..90b9aeb 100644 --- a/crates/codeoid-protocol/src/session.rs +++ b/crates/codeoid-protocol/src/session.rs @@ -46,6 +46,16 @@ pub struct SessionInfo { pub status: SessionStatus, pub created_by: String, pub created_at: String, + /// Last time this session changed state — a turn started, a tool ran, it + /// went idle. The ordering key for a relevance-sorted session list; the + /// daemon has always tracked it (it orders the resumed list by it) and now + /// puts it on the wire. + /// + /// Absent from a daemon that predates the field: fall back to `created_at` + /// rather than treating it as "never active", which would sink every + /// session to the bottom. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub last_activity_at: Option, pub attached_clients: u32, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -107,6 +117,33 @@ pub struct SessionInfo { /// which marks the orchestrating parent. #[serde(default, skip_serializing_if = "Option::is_none")] pub collaboration_role: Option, + + /// What this session IS in the fleet: the tenant's conductor, a + /// dispatch-spawned worker, or — when absent — an ordinary session. + /// + /// The daemon has carried this on the wire since the conductor shipped; + /// this crate simply never modelled it, so the TUI could not so much as + /// badge a conductor. Required by the fleet board + /// (docs/conductor-frontends-design.md §10–§11). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub role: Option, +} + +/// A session's place in the fleet. +/// +/// `#[serde(other)]` on `Unknown` is load-bearing: this is a client talking to +/// a daemon that may be NEWER than it. A future role (a domain sub-conductor, +/// say) must degrade to "some role I don't render" rather than fail to +/// deserialize the whole `SessionInfo` and blank the session list. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum SessionRole { + /// The per-tenant fleet supervisor. + Conductor, + /// A disposable worker created by a dispatch. + Worker, + #[serde(other)] + Unknown, } /// Where a forked session came from — the parent id, the parent's name at @@ -162,6 +199,21 @@ pub struct CollaborationRole { /// to. Write authority is opt-in per role. #[serde(default, skip_serializing_if = "Option::is_none")] pub write: Option, + /// Goal-blackboard artifact kinds this role may READ — `spec`, `research`, + /// `adr`, `task-list`, `diff`, `findings`, or `extra/`. + /// + /// `None` = the daemon's default profile for this role name; a role with no + /// profile and no declaration reads nothing. This is what makes reviewer + /// independence structural: `review` reads `diff`+`spec` and NOT + /// `research` or its peers' `findings`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reads: Option>, + /// Artifact kinds this role may WRITE. `None` = the default profile for + /// this role name. A role writing a multi-writer kind (`findings`) writes + /// into its own slot, chosen daemon-side, so one reviewer can never + /// overwrite another's. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub writes: Option>, } /// Set on a role-CHILD of a collaborative session: which collaboration it diff --git a/crates/codeoid-protocol/tests/wire_format.rs b/crates/codeoid-protocol/tests/wire_format.rs index e698fd6..b05e579 100644 --- a/crates/codeoid-protocol/tests/wire_format.rs +++ b/crates/codeoid-protocol/tests/wire_format.rs @@ -8,9 +8,10 @@ use codeoid_protocol::{ Attachment, CancelReason, ClientMessage, ConfirmedBy, ContentPart, DaemonMessage, ErrorCode, + FleetDelta, FleetScope, FleetTask, FleetTaskKind, FleetTaskShape, FleetTaskStatus, IdentityType, MessageIdentity, MessageRole, SearchScope, SendPriority, SessionInfo, - SessionMessage, SessionMessageDelta, SessionMode, SessionStatus, SessionUsage, ToolInfo, - ToolState, + SessionMessage, SessionMessageDelta, SessionMode, SessionRole, SessionStatus, SessionUsage, + ToolInfo, ToolState, }; use serde::Serialize; use serde_json::Value; @@ -83,6 +84,8 @@ fn sample_session_info() -> SessionInfo { created_by: "me".into(), created_at: "2026-04-22T00:00:00Z".into(), attached_clients: 1, + role: Some(SessionRole::Conductor), + last_activity_at: Some("2026-04-22T01:00:00Z".into()), mode: Some(SessionMode::Interactive), turns_remaining: Some(10), pinned_files: Some(vec!["README.md".into()]), @@ -128,6 +131,8 @@ fn sample_session_info() -> SessionInfo { count: None, purpose: None, write: None, + reads: None, + writes: None, }, codeoid_protocol::CollaborationRole { name: "review".into(), @@ -136,6 +141,8 @@ fn sample_session_info() -> SessionInfo { count: Some(3), purpose: Some("independent critique".into()), write: Some(false), + reads: Some(vec!["spec".into(), "diff".into()]), + writes: Some(vec!["findings".into()]), }, ], }), @@ -565,3 +572,121 @@ fn tool_completed_roundtrips_confirmed_by() { _ => panic!("wrong variant"), } } + +// ── Fleet board: cross-language fixtures ──────────────────────────────────── +// +// The JSON below was CAPTURED from the real TS daemon (SessionManager handling +// `fleet.subscribe`, then broadcasting a delta), not hand-written. That is the +// whole point: this crate is a hand-maintained mirror of a TS contract, and the +// failure mode is silent field drop on one side. If the daemon's projection +// changes shape, these stop deserializing. + +#[test] +fn fleet_snapshot_result_parses_real_daemon_json() { + let raw = r#"{ + "type": "fleet.snapshot.result", + "requestId": "1", + "fleet": { + "workers": [], + "tasks": [{ + "id": "1aa8e2c8-6e3c-4b85-9ffc-924ae9eda6f4", + "kind": "spawn", + "shape": "scout", + "status": "queued", + "attempts": 0, + "createdAt": 1786667059557, + "createdBy": "wimse://conductor/acc" + }], + "events": [], + "agg": { + "activeTasks": 1, "blockedTasks": 0, + "inputTokens": 0, "outputTokens": 0, "totalCostUsd": 0 + } + } + }"#; + + let msg: DaemonMessage = serde_json::from_str(raw).expect("daemon JSON must deserialize"); + let DaemonMessage::FleetSnapshotResult { request_id, fleet } = msg else { + panic!("expected FleetSnapshotResult, got {msg:?}"); + }; + assert_eq!(request_id, "1"); + assert!( + fleet.conductor.is_none(), + "absent conductor is a valid board" + ); + assert_eq!(fleet.tasks.len(), 1); + assert_eq!(fleet.tasks[0].kind, FleetTaskKind::Spawn); + assert_eq!(fleet.tasks[0].shape, FleetTaskShape::Scout); + assert_eq!(fleet.tasks[0].status, FleetTaskStatus::Queued); + assert_eq!(fleet.agg.active_tasks, 1); + // The daemon must never put the dispatch prompt on the wire, so there is + // no field here to hold one. + assert!(!raw.contains("\"prompt\"")); +} + +#[test] +fn fleet_update_delta_parses_real_daemon_json() { + let raw = r#"{ + "type": "fleet.update", + "delta": { + "kind": "task", + "task": { + "id": "4c2e690f-a7ce-40e2-baee-ad7d4998f892", + "kind": "send", + "shape": "ship", + "status": "queued", + "attempts": 0, + "createdAt": 1786667059558, + "targetSession": "sess-1", + "createdBy": "wimse://conductor/acc" + }, + "agg": { + "activeTasks": 2, "blockedTasks": 0, + "inputTokens": 0, "outputTokens": 0, "totalCostUsd": 0 + } + } + }"#; + + let msg: DaemonMessage = serde_json::from_str(raw).expect("daemon JSON must deserialize"); + let DaemonMessage::FleetUpdate { delta } = msg else { + panic!("expected FleetUpdate, got {msg:?}"); + }; + let FleetDelta::Task { task, agg } = delta else { + panic!("expected a task delta"); + }; + assert_eq!(task.kind, FleetTaskKind::Send); + assert_eq!(task.target_session.as_deref(), Some("sess-1")); + assert_eq!(agg.active_tasks, 2); +} + +#[test] +fn unknown_enum_values_degrade_instead_of_failing_the_board() { + // A client WILL meet a newer daemon. A role/status/shape it has never heard + // of must land as Unknown, not poison the whole SessionInfo or task row — + // otherwise one new enum value blanks the entire fleet view. + let task: FleetTask = serde_json::from_str( + r#"{"id":"t","kind":"teleport","shape":"warp","status":"vibing", + "attempts":0,"createdAt":1,"createdBy":"c"}"#, + ) + .expect("unknown enum values must still deserialize"); + assert_eq!(task.kind, FleetTaskKind::Unknown); + assert_eq!(task.shape, FleetTaskShape::Unknown); + assert_eq!(task.status, FleetTaskStatus::Unknown); + + let role: SessionRole = serde_json::from_str("\"sub-conductor\"").expect("unknown role"); + assert_eq!(role, SessionRole::Unknown); +} + +#[test] +fn fleet_subscribe_serializes_the_way_the_daemon_schema_demands() { + // The daemon's zod schema pins `scope` to the literal "tenant" and rejects + // anything else, so this must not serialize as, say, "Tenant". + let json = serde_json::to_value(ClientMessage::FleetSubscribe { + id: "r1".into(), + scope: FleetScope::Tenant, + }) + .unwrap(); + assert_eq!(json["type"], "fleet.subscribe"); + assert_eq!(json["scope"], "tenant"); + assert_eq!(json["id"], "r1"); +} diff --git a/crates/codeoid-tui/src/app.rs b/crates/codeoid-tui/src/app.rs index 2d941d1..0c0e2a7 100644 --- a/crates/codeoid-tui/src/app.rs +++ b/crates/codeoid-tui/src/app.rs @@ -1282,6 +1282,14 @@ impl App { ); state.provider_commands.insert(session_id, commands); } + // Fleet board frames are accepted by the protocol crate but not yet + // rendered — the board UI is P5.2. Logged rather than warned: this + // TUI never sends `fleet.subscribe`, so receiving one would mean a + // stray broadcast, not a client bug, and it must not look like a + // forward-compat drop. + DaemonMessage::FleetSnapshotResult { .. } | DaemonMessage::FleetUpdate { .. } => { + debug!("fleet board frame received; no conductor surface yet (P5.2)"); + } DaemonMessage::Unknown => { warn!("received unknown daemon message; forward-compat drop"); } @@ -2636,6 +2644,8 @@ mod tests { name: "demo".into(), workdir: "/tmp".into(), status: SessionStatus::Idle, + role: None, + last_activity_at: None, created_by: "u".into(), created_at: "t".into(), attached_clients: 0, diff --git a/crates/codeoid-tui/src/state/mod.rs b/crates/codeoid-tui/src/state/mod.rs index f94df8d..cc5c931 100644 --- a/crates/codeoid-tui/src/state/mod.rs +++ b/crates/codeoid-tui/src/state/mod.rs @@ -1139,6 +1139,8 @@ mod tests { name: "demo".into(), workdir: "/tmp".into(), status: codeoid_protocol::SessionStatus::Idle, + role: None, + last_activity_at: None, created_by: "u".into(), created_at: "2026-06-23T00:00:00Z".into(), attached_clients: 0, diff --git a/crates/codeoid-tui/src/state/sessions.rs b/crates/codeoid-tui/src/state/sessions.rs index 1ac3458..ae4979e 100644 --- a/crates/codeoid-tui/src/state/sessions.rs +++ b/crates/codeoid-tui/src/state/sessions.rs @@ -114,6 +114,8 @@ mod tests { name: name.into(), workdir: "/tmp".into(), status: SessionStatus::Idle, + role: None, + last_activity_at: None, created_by: "me".into(), created_at: "2026-04-22T00:00:00Z".into(), attached_clients: 1, diff --git a/crates/codeoid-tui/src/ui/approval.rs b/crates/codeoid-tui/src/ui/approval.rs index 51c0758..085ec31 100644 --- a/crates/codeoid-tui/src/ui/approval.rs +++ b/crates/codeoid-tui/src/ui/approval.rs @@ -276,6 +276,8 @@ mod tests { name: "demo".into(), workdir: "/tmp".into(), status: SessionStatus::WaitingApproval, + role: None, + last_activity_at: None, created_by: "u".into(), created_at: "2026-06-23T00:00:00Z".into(), attached_clients: 0, diff --git a/crates/codeoid-tui/src/ui/scrollback.rs b/crates/codeoid-tui/src/ui/scrollback.rs index fbca6dd..0b77646 100644 --- a/crates/codeoid-tui/src/ui/scrollback.rs +++ b/crates/codeoid-tui/src/ui/scrollback.rs @@ -627,6 +627,8 @@ mod tests { name: "demo".into(), workdir: "/tmp".into(), status: SessionStatus::Idle, + role: None, + last_activity_at: None, created_by: "u".into(), created_at: "2026-06-23T00:00:00Z".into(), attached_clients: 0, diff --git a/crates/codeoid-tui/src/ui/worker.rs b/crates/codeoid-tui/src/ui/worker.rs index 51c539e..f467709 100644 --- a/crates/codeoid-tui/src/ui/worker.rs +++ b/crates/codeoid-tui/src/ui/worker.rs @@ -424,6 +424,8 @@ mod tests { name: "demo".into(), workdir: "/tmp".into(), status: SessionStatus::Idle, + role: None, + last_activity_at: None, created_by: "u".into(), created_at: "t".into(), attached_clients: 0,