refactor(dispatch): SessionDriver seam + protocol v4 — close the session-parity gaps structurally

src/apps/cli/src/dispatch/mod.rs

@@ -17,10 +17,12 @@ use protocol::{
     DispatchAnswerRequest, DispatchAnswerResponse, DispatchAppendRequest, DispatchAppendResponse,
     DispatchCancelRequest, DispatchCancelResponse, DispatchContinueRequest,
     DispatchContinueResponse, DispatchJobListEntry, DispatchJobState, DispatchListRequest,
-    DispatchProbeRequest, DispatchProbeResponse, DispatchStatusRequest, DispatchStatusResponse,
-    DispatchSubmitRequest, DispatchSubmitResponse, DispatchWorkspaceBundleBeginRequest,
-    DispatchWorkspaceBundleChunkRequest, DispatchWorkspaceBundleCommitRequest,
-    DispatchWorkspaceProbe, DispatchWorkspaceProvisionRequest, DispatchWorkspaceSyncChunkRequest,
+    DispatchProbeRequest, DispatchProbeResponse, DispatchQueryKind, DispatchQueryRequest,
+    DispatchStatusRequest, DispatchStatusResponse,
+    DispatchSubmitRequest, DispatchSubmitResponse, DispatchTurnKind,
+    DispatchWorkspaceBundleBeginRequest, DispatchWorkspaceBundleChunkRequest,
+    DispatchWorkspaceBundleCommitRequest, DispatchWorkspaceProbe,
+    DispatchWorkspaceProvisionRequest, DispatchWorkspaceSyncChunkRequest,
     DispatchWorkspaceSyncRequest, DISPATCH_PROTOCOL_VERSION, MAX_DISPATCH_TEXT_BYTES,
 };
 use store::{CreateJobOutcome, DispatchStateRecord, DispatchStore};
@@ -63,6 +65,7 @@ pub(crate) async fn run_dispatch_verb(
         "append" => serde_json::to_value(append(parse(input)?)?).context("encode appended message"),
         "continue" => serde_json::to_value(continue_job(parse(input)?)?)
             .context("encode follow-up turn response"),
+        "query" => query(parse(input)?).await.context("encode query response"),
         "workspace-provision" => serde_json::to_value(workspace::provision(parse::<
             DispatchWorkspaceProvisionRequest,
         >(input)?)?)
@@ -120,30 +123,16 @@ async fn probe(request: DispatchProbeRequest) -> Result<DispatchProbeResponse> {
         .as_deref()
         .map(inspect_workspace)
         .transpose()?;
-    let mut capabilities = vec![
-        "persistent_jobs".to_string(),
-        "cursor_events".to_string(),
-        "workspace_serialization".to_string(),
-        "approval_auto".to_string(),
-        "approval_reject_and_report".to_string(),
-        "approval_remote".to_string(),
-        "frontend_event_projection".to_string(),
-        "append_message".to_string(),
-        "event_log_completeness".to_string(),
-        // Git-worktree delivery. A target without these cannot be provisioned
-        // at all — there is no snapshot fallback left — so controllers fail
-        // preflight rather than degrade.
-        "workspace_git_worktree".to_string(),
-        "workspace_git_bundle_upload".to_string(),
-        "workspace_git_sync".to_string(),
-        // A target may share the same package version while predating the
-        // dispatch entrypoint's early CLI-profile selection. Such a binary can
-        // accept a job but every detached worker then fails before execution.
-        // Advertise the behavioral fix explicitly so controllers fail closed.
-        "dispatch_worker_cli_profile".to_string(),
-    ];
+    let mut capabilities: Vec<String> =
+        bitfun_services_core::dispatch_contract::DISPATCH_BASE_TARGET_CAPABILITIES
+            .iter()
+            .map(|capability| capability.to_string())
+            .collect();
     if runner::is_supported() {
-        capabilities.push("detached_worker".to_string());
+        capabilities.push(
+            bitfun_services_core::dispatch_contract::DISPATCH_DETACHED_WORKER_CAPABILITY
+                .to_string(),
+        );
     }
     Ok(DispatchProbeResponse {
         protocol_version: DISPATCH_PROTOCOL_VERSION,
@@ -225,12 +214,33 @@ fn continue_job(request: DispatchContinueRequest) -> Result<DispatchContinueResp
             DISPATCH_PROTOCOL_VERSION
         );
     }
-    if request.prompt.trim().is_empty() {
-        bail!("dispatch follow-up requires a prompt");
+    match request.kind {
+        DispatchTurnKind::Prompt => {
+            if request.prompt.trim().is_empty() {
+                bail!("dispatch follow-up requires a prompt");
+            }
+        }
+        DispatchTurnKind::Compact => {
+            if !request.prompt.trim().is_empty() {
+                bail!("dispatch compact turns do not take a prompt");
+            }
+            if !request.attachments.is_empty() {
+                bail!("dispatch compact turns do not take attachments");
+            }
+        }
     }
+    validate_attachments(&request.attachments)?;
     if request.prompt.len() > MAX_DISPATCH_TEXT_BYTES {
         bail!("dispatch follow-up prompt exceeds the 32 KiB safety limit");
     }
+    if let Some(model) = &request.model {
+        if model.trim().is_empty() {
+            bail!("dispatch model override cannot be empty");
+        }
+        if model.len() > 256 {
+            bail!("dispatch model override exceeds the 256 byte limit");
+        }
+    }
     if !runner::is_supported() {
         bail!("dispatch detached workers are supported only on Linux and macOS");
     }
@@ -249,6 +259,45 @@ fn continue_job(request: DispatchContinueRequest) -> Result<DispatchContinueResp
     })
 }
 
+/// Answer a read-only session question from persisted state.
+///
+/// Deliberately runtime-free: `PersistenceManager` reads the session's
+/// on-disk turns directly, so the query can run while a detached worker owns
+/// the live session without contending for anything.
+async fn query(request: DispatchQueryRequest) -> Result<serde_json::Value> {
+    let store = DispatchStore::open_default()?;
+    let job = store.load_job(&request.job_id)?;
+    match request.kind {
+        DispatchQueryKind::UsageReport => {
+            let path_manager = bitfun_core::infrastructure::PathManager::new()
+                .map_err(|error| anyhow::anyhow!("resolve BitFun storage root: {error}"))?;
+            let persistence = bitfun_core::agentic::persistence::PersistenceManager::new(
+                std::sync::Arc::new(path_manager),
+            )
+            .map_err(|error| anyhow::anyhow!("open session persistence: {error}"))?;
+            let report = bitfun_core::service::session_usage::generate_session_usage_report(
+                &persistence,
+                None,
+                bitfun_core::service::session_usage::SessionUsageReportRequest {
+                    session_id: job.request.session_id.clone(),
+                    workspace_path: Some(job.request.workspace_path.clone()),
+                    remote_connection_id: None,
+                    remote_ssh_host: None,
+                    include_hidden_subagents: false,
+                },
+            )
+            .await
+            .map_err(|error| anyhow::anyhow!("generate dispatch usage report: {error}"))?;
+            serde_json::to_value(serde_json::json!({
+                "kind": "usageReport",
+                "sessionId": job.request.session_id,
+                "report": report,
+            }))
+            .context("encode dispatch usage report")
+        }
+    }
+}
+
 fn ensure_worker_spawned(
     store: &DispatchStore,
     job_id: &str,
@@ -706,6 +755,10 @@ fn canonical_workspace(workspace_path: &str) -> Result<PathBuf> {
     Ok(canonical)
 }
 
+fn validate_attachments(attachments: &[protocol::DispatchAttachment]) -> Result<()> {
+    protocol::validate_dispatch_attachments(attachments).map_err(anyhow::Error::msg)
+}
+
 fn validate_submit_request(request: &DispatchSubmitRequest) -> Result<()> {
     if request.protocol_version != DISPATCH_PROTOCOL_VERSION {
         bail!(
@@ -729,6 +782,7 @@ fn validate_submit_request(request: &DispatchSubmitRequest) -> Result<()> {
     if request.prompt.len() > MAX_DISPATCH_TEXT_BYTES {
         bail!("dispatch prompt exceeds the 32 KiB request limit");
     }
+    validate_attachments(&request.attachments)?;
     if request.setup_audit.len() > 32 {
         bail!("dispatch setup audit exceeds the 32-event safety limit");
     }
@@ -775,6 +829,7 @@ mod tests {
             approval_policy: DispatchApprovalPolicy::RejectAndReport,
             model: Some("model-1".to_string()),
             title: Some("Task".to_string()),
+            attachments: Vec::new(),
             setup_audit: Vec::new(),
         }
     }

src/apps/cli/src/dispatch/protocol.rs

@@ -2,7 +2,12 @@ use serde::{Deserialize, Serialize};
 
 use bitfun_agent_runtime::sdk::{PermissionReply, PermissionRequest};
 
-pub(crate) const DISPATCH_PROTOCOL_VERSION: u32 = 3;
+// The wire contract (version, capability names, attachment shape and
+// limits) has one source of truth shared with the controller side.
+pub(crate) use bitfun_services_core::dispatch_contract::{
+    validate_dispatch_attachments, DispatchAttachment, DISPATCH_PROTOCOL_VERSION,
+};
+
 pub(crate) const MAX_DISPATCH_TEXT_BYTES: usize = 32 * 1024;
 
 #[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq)]
@@ -55,6 +60,17 @@ pub(crate) enum DispatchApprovalPolicy {
     Remote,
 }
 
+/// What a queued follow-up turn asks the worker to run.
+#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub(crate) enum DispatchTurnKind {
+    /// An ordinary user prompt submitted as a dialog turn.
+    #[default]
+    Prompt,
+    /// Manual context compaction, run as a turn so its events attribute.
+    Compact,
+}
+
 #[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub(crate) struct DispatchSubmitRequest {
@@ -69,6 +85,8 @@ pub(crate) struct DispatchSubmitRequest {
     pub(crate) model: Option<String>,
     #[serde(default)]
     pub(crate) title: Option<String>,
+    #[serde(default)]
+    pub(crate) attachments: Vec<DispatchAttachment>,
     /// Controller-side setup actions that happened before the target job could
     /// exist (currently the signed CLI auto-install). They are replayed into
     /// the durable job event log at creation time and are deliberately excluded
@@ -338,6 +356,18 @@ pub(crate) struct DispatchContinueRequest {
     pub(crate) prompt: String,
     #[serde(default)]
     pub(crate) display_content: Option<String>,
+    /// Per-turn model override. Absent keeps the job's current model; present
+    /// it also becomes the job's model for later turns.
+    #[serde(default)]
+    pub(crate) model: Option<String>,
+    /// Per-turn approval-policy override with the same carry-forward rule.
+    #[serde(default)]
+    pub(crate) approval_policy: Option<DispatchApprovalPolicy>,
+    /// Operation the worker runs; defaults to an ordinary prompt turn.
+    #[serde(default)]
+    pub(crate) kind: DispatchTurnKind,
+    #[serde(default)]
+    pub(crate) attachments: Vec<DispatchAttachment>,
 }
 
 #[derive(Clone, Debug, Serialize, PartialEq, Eq)]
@@ -356,6 +386,24 @@ pub(crate) struct DispatchCancelRequest {
     pub(crate) job_id: String,
 }
 
+/// Read-only question about a job's persisted session state.
+///
+/// Served by a short-lived process straight from persistence — no runtime is
+/// initialized and no workspace runtime ownership is taken, so a query is
+/// always safe next to a running detached worker.
+#[derive(Clone, Debug, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase", deny_unknown_fields)]
+pub(crate) struct DispatchQueryRequest {
+    pub(crate) job_id: String,
+    pub(crate) kind: DispatchQueryKind,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Serialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub(crate) enum DispatchQueryKind {
+    UsageReport,
+}
+
 #[derive(Clone, Debug, Default, Deserialize, PartialEq, Eq)]
 #[serde(rename_all = "camelCase", deny_unknown_fields)]
 pub(crate) struct DispatchListRequest {}
@@ -406,6 +454,14 @@ impl DispatchEvent {
         }
     }
 
+    pub(crate) fn model_selected(model: Option<&str>) -> Self {
+        Self::Audit {
+            timestamp: chrono::Utc::now().to_rfc3339(),
+            action: "modelSelected".to_string(),
+            details: serde_json::json!({ "model": model }),
+        }
+    }
+
     pub(crate) fn cancel_requested() -> Self {
         Self::Audit {
             timestamp: chrono::Utc::now().to_rfc3339(),