11 Commits
31 changed files with 2818 additions and 398 deletions
Generated
+1
View File
@@ -6067,6 +6067,7 @@ dependencies = [
"config-source",
"dotenv",
"flow",
"fs-operation",
"fs4",
"futures",
"futures-util",
+25
View File
@@ -214,6 +214,31 @@ impl Scope {
})
}
/// Resolve one rule target with the same symlink and missing-tail semantics
/// used by scope matching.
pub fn resolved_target(rule: &ScopeRule) -> Result<PathBuf, ScopeError> {
Ok(resolve_rule(rule)?.target)
}
/// Return whether this effective scope fully contains a requested rule.
/// This is used when attenuating provider authority without mutating the
/// parent scope.
pub fn allows_rule(&self, requested: &ScopeRule) -> Result<bool, ScopeError> {
let requested = resolve_rule(requested)?;
let covered = self
.allow
.iter()
.any(|candidate| rule_covers(candidate, &requested));
if !covered {
return Ok(false);
}
let denied = self
.deny
.iter()
.any(|deny| denial_overlaps_requested(deny, &requested));
Ok(!denied)
}
/// Effective permission for `path`.
///
/// Returns `None` when `path` is outside every allow rule, or when
+110
View File
@@ -278,6 +278,47 @@ impl Method {
// Event (Worker → Client via Unix Socket broadcast)
// ---------------------------------------------------------------------------
/// Presentation category for an Internal Worker exposed through its parent's
/// protocol stream. Internal Workers never become independently addressable
/// protocol subjects.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(rename_all = "snake_case")]
pub enum InternalWorkerKind {
SubWorker,
}
/// Stable presentation identity for one parent-owned Internal Worker session.
/// `name` is display-only; `session_id` is the identity used by clients.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct InternalWorkerRef {
pub session_id: String,
pub name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub parent_session_id: Option<String>,
pub kind: InternalWorkerKind,
}
/// Reconnect state for one visible Internal Worker. The revision fences live
/// `Event::InternalWorker` updates that raced with parent snapshot assembly.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
pub struct InternalWorkerSnapshot {
pub worker: InternalWorkerRef,
pub revision: u64,
#[cfg_attr(feature = "typescript", ts(type = "Array<unknown>"))]
pub entries: Vec<serde_json::Value>,
#[serde(default)]
pub status: WorkerStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub error: Option<String>,
#[serde(default, skip_serializing_if = "InFlightSnapshot::is_empty")]
pub in_flight: InFlightSnapshot,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub internal_workers: Vec<InternalWorkerSnapshot>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
#[serde(tag = "event", content = "data", rename_all = "snake_case")]
@@ -476,6 +517,18 @@ pub enum Event {
/// run but is not yet represented by committed snapshot entries.
#[serde(default, skip_serializing_if = "InFlightSnapshot::is_empty")]
in_flight: InFlightSnapshot,
/// Parent-owned Internal Worker sessions visible to this client.
/// Service-private Internal Workers are deliberately excluded.
#[serde(default, skip_serializing_if = "Vec::is_empty")]
internal_workers: Vec<InternalWorkerSnapshot>,
},
/// A live event from a parent-owned Internal Worker. The payload reuses the
/// normal Worker event vocabulary while the wrapper carries stable origin
/// identity and a per-child revision fence.
InternalWorker {
worker: InternalWorkerRef,
revision: u64,
event: Box<Event>,
},
/// Server-side segment log rotated to a fresh `SegmentStart`.
///
@@ -1269,6 +1322,7 @@ mod tests {
},
status: WorkerStatus::Paused,
in_flight: InFlightSnapshot::default(),
internal_workers: Vec::new(),
};
let json = serde_json::to_string(&event).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
@@ -1322,6 +1376,7 @@ mod tests {
},
],
},
internal_workers: Vec::new(),
};
let json = serde_json::to_string(&event).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
@@ -1716,6 +1771,61 @@ mod tests {
}
}
#[test]
fn internal_worker_event_roundtrip_preserves_origin_and_payload() {
let event = Event::InternalWorker {
worker: InternalWorkerRef {
session_id: "session-1".into(),
name: "research".into(),
parent_session_id: Some("parent-session".into()),
kind: InternalWorkerKind::SubWorker,
},
revision: 7,
event: Box::new(Event::TextDone {
text: "result".into(),
}),
};
let json = serde_json::to_string(&event).unwrap();
let decoded: Event = serde_json::from_str(&json).unwrap();
match decoded {
Event::InternalWorker {
worker,
revision,
event,
} => {
assert_eq!(worker.session_id, "session-1");
assert_eq!(worker.parent_session_id.as_deref(), Some("parent-session"));
assert_eq!(revision, 7);
assert!(matches!(*event, Event::TextDone { ref text } if text == "result"));
}
other => panic!("expected internal Worker event, got {other:?}"),
}
}
#[test]
fn legacy_snapshot_defaults_internal_workers_to_empty() {
let snapshot: Event = serde_json::from_value(serde_json::json!({
"event": "snapshot",
"data": {
"entries": [],
"greeting": {
"worker_name": "parent",
"cwd": ".",
"provider": "provider",
"model": "model",
"scope_summary": "scope",
"tools": []
},
"status": "idle"
}
}))
.unwrap();
assert!(matches!(
snapshot,
Event::Snapshot { internal_workers, .. } if internal_workers.is_empty()
));
}
#[test]
fn worker_discovery_events_roundtrip() {
let events = [
+7 -3
View File
@@ -4,9 +4,10 @@ use ts_rs::{Config, TS};
use crate::{
Alert, AlertLevel, AlertSource, CompletionEntry, CompletionKind, ErrorCode, Event, Greeting,
InFlightBlock, InFlightSnapshot, InFlightToolCallState, InvokeKind, MemoryWorkerEvent, Method,
Permission, RewindSummary, RewindTarget, RewindTargetId, RunResult, ScopeRule, Segment,
TurnResult, WorkerEvent, WorkerStatus,
InFlightBlock, InFlightSnapshot, InFlightToolCallState, InternalWorkerKind, InternalWorkerRef,
InternalWorkerSnapshot, InvokeKind, MemoryWorkerEvent, Method, Permission, RewindSummary,
RewindTarget, RewindTargetId, RunResult, ScopeRule, Segment, TurnResult, WorkerEvent,
WorkerStatus,
subscription::{
EventSubscriptionSelector, SubscriptionEvent, SubscriptionEventPayload, SubscriptionFrame,
SubscriptionFramePayload, SubscriptionId, SubscriptionRejectionCode, SubscriptionRequest,
@@ -53,6 +54,9 @@ pub fn generated_protocol_types() -> String {
push_decl::<RewindSummary>(&cfg, &mut output);
push_decl::<InFlightBlock>(&cfg, &mut output);
push_decl::<InFlightSnapshot>(&cfg, &mut output);
push_decl::<InternalWorkerKind>(&cfg, &mut output);
push_decl::<InternalWorkerRef>(&cfg, &mut output);
push_decl::<InternalWorkerSnapshot>(&cfg, &mut output);
push_decl::<Greeting>(&cfg, &mut output);
push_decl::<Alert>(&cfg, &mut output);
push_decl::<MemoryWorkerEvent>(&cfg, &mut output);
+155 -2
View File
@@ -4,8 +4,8 @@ use std::time::{Duration, Instant};
use protocol::{
AlertLevel, AlertSource, CompletionEntry, CompletionKind, ErrorCode, Event, InFlightBlock,
InFlightSnapshot, InFlightToolCallState, Method, RewindTarget, RunResult, Segment,
WorkerStatus,
InFlightSnapshot, InFlightToolCallState, InternalWorkerRef, InternalWorkerSnapshot, Method,
RewindTarget, RunResult, Segment, WorkerStatus,
};
use crate::block::{
@@ -227,6 +227,12 @@ impl ActionbarNotice {
}
}
pub struct InternalWorkerView {
pub worker: InternalWorkerRef,
pub revision: u64,
pub app: Box<App>,
}
pub struct App {
pub worker_name: String,
pub connected: bool,
@@ -274,6 +280,9 @@ pub struct App {
/// Turn/protocol errors retained when a real `SegmentStart` replaces the
/// replayable conversation rows during segment rotation.
run_error_messages: Vec<String>,
/// Presentation-only Internal Worker projections keyed by session identity.
/// They are rendered in separate sub-panes and never mixed into `blocks`.
pub internal_workers: Vec<InternalWorkerView>,
pub scroll: Scroll,
pub mode: Mode,
pub cache: FileCache,
@@ -351,6 +360,7 @@ impl App {
quit_confirm: None,
blocks: Vec::new(),
run_error_messages: Vec::new(),
internal_workers: Vec::new(),
scroll: Scroll::default(),
mode: Mode::Normal,
cache: FileCache::new(),
@@ -1296,11 +1306,18 @@ impl App {
greeting,
status,
in_flight,
internal_workers,
} => {
self.rewind_refresh_fence = false;
self.restore_snapshot(&entries, greeting, in_flight);
self.replace_internal_worker_snapshots(internal_workers);
self.set_worker_status(status);
}
Event::InternalWorker {
worker,
revision,
event,
} => self.apply_internal_worker_event(worker, revision, *event),
Event::Status { status } => {
self.rewind_refresh_fence = false;
self.set_worker_status(status);
@@ -1980,6 +1997,60 @@ impl App {
/// LogEntry variant into the same blocks live events would have
/// produced. Followed by `Event::Entry` updates for anything
/// committed after the snapshot.
fn replace_internal_worker_snapshots(&mut self, snapshots: Vec<InternalWorkerSnapshot>) {
self.internal_workers = snapshots
.into_iter()
.map(Self::internal_worker_view_from_snapshot)
.collect();
}
fn internal_worker_view_from_snapshot(snapshot: InternalWorkerSnapshot) -> InternalWorkerView {
let mut app = App::new(snapshot.worker.name.clone());
app.restore_entries(&snapshot.entries, None);
app.apply_in_flight_snapshot(snapshot.in_flight);
app.set_worker_status(snapshot.status);
if let Some(error) = snapshot.error {
let _ = app.handle_worker_event(Event::Error {
code: protocol::ErrorCode::Internal,
message: error,
});
}
app.replace_internal_worker_snapshots(snapshot.internal_workers);
InternalWorkerView {
worker: snapshot.worker,
revision: snapshot.revision,
app: Box::new(app),
}
}
fn apply_internal_worker_event(
&mut self,
worker: InternalWorkerRef,
revision: u64,
event: Event,
) {
let index = self
.internal_workers
.iter()
.position(|candidate| candidate.worker.session_id == worker.session_id);
let target = if let Some(index) = index {
&mut self.internal_workers[index]
} else {
self.internal_workers.push(InternalWorkerView {
worker: worker.clone(),
revision: 0,
app: Box::new(App::new(worker.name.clone())),
});
self.internal_workers.last_mut().unwrap()
};
if revision <= target.revision {
return;
}
target.worker = worker;
target.revision = revision;
let _ = target.app.handle_worker_event(event);
}
fn restore_snapshot(
&mut self,
entries: &[serde_json::Value],
@@ -3276,6 +3347,7 @@ mod completion_flow_tests {
entries: vec![session_start_value],
status: WorkerStatus::Running,
in_flight: Default::default(),
internal_workers: Vec::new(),
});
assert!(matches!(app.worker_status, WorkerStatus::Running));
@@ -3321,6 +3393,7 @@ mod completion_flow_tests {
entries: vec![serde_json::to_value(run_errored).unwrap()],
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
});
let errors = app
@@ -3398,6 +3471,7 @@ mod completion_flow_tests {
},
],
},
internal_workers: Vec::new(),
});
app.handle_worker_event(Event::TextDelta { text: "lo".into() });
@@ -3434,6 +3508,83 @@ mod completion_flow_tests {
assert!(app.blocks.is_empty());
}
#[test]
fn internal_worker_events_project_by_session_without_mixing_parent_blocks() {
let mut app = App::new("parent".into());
let worker = InternalWorkerRef {
session_id: "child-session".into(),
name: "research".into(),
parent_session_id: Some("parent-session".into()),
kind: protocol::InternalWorkerKind::SubWorker,
};
app.handle_worker_event(Event::InternalWorker {
worker: worker.clone(),
revision: 2,
event: Box::new(Event::TextDelta {
text: "child output".into(),
}),
});
app.handle_worker_event(Event::InternalWorker {
worker,
revision: 1,
event: Box::new(Event::TextDelta {
text: "stale".into(),
}),
});
assert!(app.blocks.is_empty());
assert_eq!(app.internal_workers.len(), 1);
assert_eq!(app.internal_workers[0].revision, 2);
assert!(
app.internal_workers[0].app.blocks.iter().any(
|block| matches!(block, Block::AssistantText { text } if text == "child output")
)
);
}
#[test]
fn snapshot_authoritatively_replaces_internal_worker_views() {
let mut app = App::new("parent".into());
app.internal_workers.push(InternalWorkerView {
worker: InternalWorkerRef {
session_id: "old".into(),
name: "old".into(),
parent_session_id: None,
kind: protocol::InternalWorkerKind::SubWorker,
},
revision: 1,
app: Box::new(App::new("old".into())),
});
app.handle_worker_event(Event::Snapshot {
greeting: test_greeting(),
entries: Vec::new(),
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: vec![InternalWorkerSnapshot {
worker: InternalWorkerRef {
session_id: "replacement".into(),
name: "replacement".into(),
parent_session_id: Some("parent-session".into()),
kind: protocol::InternalWorkerKind::SubWorker,
},
revision: 4,
entries: Vec::new(),
status: WorkerStatus::Running,
error: None,
in_flight: Default::default(),
internal_workers: Vec::new(),
}],
});
assert_eq!(app.internal_workers.len(), 1);
assert_eq!(app.internal_workers[0].worker.session_id, "replacement");
assert_eq!(app.internal_workers[0].revision, 4);
assert_eq!(
app.internal_workers[0].app.worker_status,
WorkerStatus::Running
);
}
#[test]
fn live_system_item_notification_appends_notify_block() {
let mut app = App::new("test".into());
@@ -3552,6 +3703,7 @@ mod completion_flow_tests {
greeting,
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
});
assert_eq!(app.context_window, 123_000);
@@ -3749,6 +3901,7 @@ mod completion_flow_tests {
entries: assistant_item_entries,
status: WorkerStatus::Running,
in_flight: Default::default(),
internal_workers: Vec::new(),
});
let tasks = app.task_store.tasks();
+2
View File
@@ -2019,6 +2019,7 @@ mod tests {
entries: vec![],
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
});
app.handle_worker_event(Event::RewindApplied {
entries: vec![],
@@ -2045,6 +2046,7 @@ mod tests {
entries: vec![],
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
});
type_keys(&mut app, "draft");
+2
View File
@@ -859,6 +859,7 @@ async fn ticket_queue_notification_sends_notify_when_socket_available() {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await
.unwrap();
@@ -900,6 +901,7 @@ async fn send_notify_only_can_deliver_weak_notification_without_auto_run() {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await
.unwrap();
+22
View File
@@ -381,6 +381,28 @@ pub fn compute_history(app: &App, width: u16) -> HistoryLayout {
i += 1;
}
for internal in &app.internal_workers {
logical.push((Line::from(""), false));
logical.push((
Line::from(vec![
Span::styled("SubWorker ", Style::default().bold()),
Span::raw(internal.worker.name.clone()),
Span::styled(
format!(" {:?}", internal.app.worker_status),
Style::default().fg(Color::DarkGray),
),
]),
false,
));
let child_width = width.saturating_sub(2).max(1);
let child_history = compute_history(&internal.app, child_width);
logical.extend(child_history.rows.into_iter().map(|row| {
let mut spans = vec![Span::raw(" ")];
spans.extend(row.line.spans);
(Line::from(spans), row.selectable)
}));
}
// Step 2: pre-wrap every logical line to char-based terminal rows so
// scroll math is exact. Track the logical → wrapped mapping so
// turn-start indices get translated into wrapped-row coordinates.
+1
View File
@@ -914,6 +914,7 @@ mod tests {
greeting: test_greeting(),
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
},
];
+978
View File
@@ -0,0 +1,978 @@
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, Weak};
use async_trait::async_trait;
use fs_operation::{
EditRequest, EditResult, FsPath, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest,
ListResult, ReadRequest, ReadResult, StatRequest, StatResult, WriteRequest, WriteResult,
};
use crate::{
CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, Workdir,
WorkdirError, WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability,
WorkdirSessionHandle,
};
#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkdirDelegationPermission {
Read,
Write,
}
#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkdirDelegationRule {
pub target: FsPath,
pub permission: WorkdirDelegationPermission,
pub recursive: bool,
}
#[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkdirDelegationRequest {
pub rules: Vec<WorkdirDelegationRule>,
pub cwd: FsPath,
}
pub struct WorkdirDelegation {
pub scoped_session: WorkdirSessionHandle,
pub capabilities: WorkdirSessionCapabilities,
validity: Arc<SessionValidity>,
}
impl std::fmt::Debug for WorkdirDelegation {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("WorkdirDelegation")
.field("workdir", &self.scoped_session.workdir())
.field("capabilities", &self.capabilities)
.field("active", &self.is_active())
.finish()
}
}
impl WorkdirDelegation {
pub fn is_active(&self) -> bool {
self.validity.is_active()
}
pub fn release(&self) {
self.validity.active.store(false, Ordering::Release);
}
}
impl Drop for WorkdirDelegation {
fn drop(&mut self) {
self.release();
}
}
pub struct AppliedWorkdirDelegation {
pub scoped_session: WorkdirSessionHandle,
_leases: Vec<WorkdirDelegation>,
}
impl std::fmt::Debug for AppliedWorkdirDelegation {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AppliedWorkdirDelegation")
.field("workdir", self.scoped_session.workdir())
.field("lease_count", &self._leases.len())
.finish()
}
}
pub async fn apply_delegation_chain(
source: WorkdirSessionHandle,
requests: impl IntoIterator<Item = WorkdirDelegationRequest>,
) -> Result<AppliedWorkdirDelegation, WorkdirError> {
let mut current = source;
let mut leases = Vec::new();
for request in requests {
let authority = if current.is_delegation_capable() {
current.clone()
} else {
delegation_capable_session(current.clone())
};
let lease = authority.delegate(request).await?;
current = lease.scoped_session.clone();
leases.push(lease);
}
Ok(AppliedWorkdirDelegation {
scoped_session: current,
_leases: leases,
})
}
#[derive(Debug)]
struct SessionValidity {
active: AtomicBool,
parent: Option<Arc<SessionValidity>>,
}
impl SessionValidity {
fn root() -> Arc<Self> {
Arc::new(Self {
active: AtomicBool::new(true),
parent: None,
})
}
fn child(parent: Arc<Self>) -> Arc<Self> {
Arc::new(Self {
active: AtomicBool::new(true),
parent: Some(parent),
})
}
fn is_active(&self) -> bool {
self.active.load(Ordering::Acquire)
&& self.parent.as_ref().is_none_or(|parent| parent.is_active())
}
}
#[derive(Clone, Debug)]
struct ActiveWriteLease {
validity: Weak<SessionValidity>,
rules: Vec<WorkdirDelegationRule>,
}
struct DelegatingWorkdirSession {
source: WorkdirSessionHandle,
cwd: FsPath,
scope: Option<Vec<WorkdirDelegationRule>>,
capabilities: WorkdirSessionCapabilities,
validity: Arc<SessionValidity>,
child_write_leases: Mutex<HashMap<u64, ActiveWriteLease>>,
next_lease_id: AtomicU64,
closes_source: bool,
}
impl std::fmt::Debug for DelegatingWorkdirSession {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DelegatingWorkdirSession")
.field("workdir", &self.source.workdir())
.field("scope", &self.scope)
.field("capabilities", &self.capabilities)
.field("active", &self.validity.is_active())
.finish_non_exhaustive()
}
}
/// Wrap a provider session with logical-path delegation and parent write gates.
pub fn delegation_capable_session(source: WorkdirSessionHandle) -> WorkdirSessionHandle {
let capabilities = source.capabilities();
Arc::new(DelegatingWorkdirSession {
source,
cwd: FsPath::new("").expect("empty Workdir path is valid"),
scope: None,
capabilities,
validity: SessionValidity::root(),
child_write_leases: Mutex::new(HashMap::new()),
next_lease_id: AtomicU64::new(1),
closes_source: true,
})
}
impl DelegatingWorkdirSession {
fn ensure_active(&self) -> Result<(), WorkdirError> {
if self.validity.is_active() {
Ok(())
} else {
Err(WorkdirError::SessionClosed)
}
}
fn ensure_capability(
&self,
required: WorkdirSessionCapability,
operation: &'static str,
) -> Result<(), WorkdirError> {
self.ensure_active()?;
if self.capabilities.supports(required) {
Ok(())
} else {
Err(WorkdirError::Denied(format!(
"delegated workdir session does not permit {operation}"
)))
}
}
fn ensure_path(
&self,
path: &FsPath,
permission: WorkdirDelegationPermission,
) -> Result<(), WorkdirError> {
self.ensure_active()?;
if let Some(scope) = &self.scope {
if !scope
.iter()
.any(|rule| rule_allows_path(rule, path, permission))
{
return Err(WorkdirError::Denied(format!(
"logical workdir path `{path}` is outside the delegated {permission:?} scope"
)));
}
}
if permission == WorkdirDelegationPermission::Write {
self.ensure_parent_write_available(path)?;
}
Ok(())
}
fn resolve_path(&self, path: &FsPath) -> Result<FsPath, WorkdirError> {
if self.cwd.as_str().is_empty() {
return Ok(path.clone());
}
let joined = Path::new(self.cwd.as_str()).join(path.as_str());
let joined = joined.to_str().ok_or_else(|| {
WorkdirError::Denied("logical Workdir path is not valid UTF-8".into())
})?;
FsPath::new(joined).map_err(|error| WorkdirError::Denied(error.to_string()))
}
fn ensure_read(
&self,
path: &FsPath,
capability: WorkdirSessionCapability,
) -> Result<(), WorkdirError> {
self.ensure_capability(capability, "read operations")?;
self.ensure_path(path, WorkdirDelegationPermission::Read)
}
fn ensure_write(
&self,
path: &FsPath,
capability: WorkdirSessionCapability,
) -> Result<(), WorkdirError> {
self.ensure_capability(capability, "write operations")?;
self.ensure_path(path, WorkdirDelegationPermission::Write)
}
fn ensure_command(&self, starting: bool) -> Result<(), WorkdirError> {
self.ensure_capability(WorkdirSessionCapability::Command, "command execution")?;
if starting && self.has_active_write_lease() {
return Err(WorkdirError::Denied(
"command execution is denied while a child holds a write delegation".into(),
));
}
Ok(())
}
fn ensure_parent_write_available(&self, path: &FsPath) -> Result<(), WorkdirError> {
let mut leases = self
.child_write_leases
.lock()
.expect("workdir delegation lease mutex poisoned");
leases.retain(|_, lease| lease.validity.upgrade().is_some_and(|v| v.is_active()));
if leases.values().any(|lease| {
lease.rules.iter().any(|rule| {
rule.permission == WorkdirDelegationPermission::Write
&& rule_allows_path(rule, path, WorkdirDelegationPermission::Write)
})
}) {
Err(WorkdirError::Denied(format!(
"logical workdir path `{path}` is leased to a child session"
)))
} else {
Ok(())
}
}
fn has_active_write_lease(&self) -> bool {
let mut leases = self
.child_write_leases
.lock()
.expect("workdir delegation lease mutex poisoned");
leases.retain(|_, lease| lease.validity.upgrade().is_some_and(|v| v.is_active()));
leases.values().any(|lease| {
lease
.rules
.iter()
.any(|rule| rule.permission == WorkdirDelegationPermission::Write)
})
}
fn validate_delegation_rules(
&self,
rules: &[WorkdirDelegationRule],
) -> Result<WorkdirSessionCapabilities, WorkdirError> {
self.ensure_active()?;
if rules.is_empty() {
return Err(WorkdirError::Denied(
"workdir delegation requires at least one logical scope rule".into(),
));
}
let writable = rules
.iter()
.any(|rule| rule.permission == WorkdirDelegationPermission::Write);
if !self.capabilities.supports(WorkdirSessionCapability::Read)
|| (writable
&& (!self.capabilities.supports(WorkdirSessionCapability::Write)
|| !self.capabilities.supports(WorkdirSessionCapability::Edit)))
{
return Err(WorkdirError::Denied(
"parent workdir session cannot delegate the requested capabilities".into(),
));
}
for requested in rules {
if let Some(scope) = &self.scope {
if !scope
.iter()
.any(|parent| rule_contains_rule(parent, requested))
{
return Err(WorkdirError::Denied(format!(
"logical workdir scope `{}` exceeds the parent delegation",
requested.target
)));
}
}
}
let mut delegated = vec![WorkdirSessionCapability::Read];
for capability in [
WorkdirSessionCapability::Glob,
WorkdirSessionCapability::Grep,
] {
if self.capabilities.supports(capability) {
delegated.push(capability);
}
}
if writable {
delegated.push(WorkdirSessionCapability::Write);
delegated.push(WorkdirSessionCapability::Edit);
}
Ok(WorkdirSessionCapabilities::from_capabilities(delegated))
}
}
#[async_trait]
impl WorkdirSession for DelegatingWorkdirSession {
fn workdir(&self) -> &Workdir {
self.source.workdir()
}
fn capabilities(&self) -> WorkdirSessionCapabilities {
self.capabilities
}
fn is_delegation_capable(&self) -> bool {
true
}
fn transports_delegation_context(&self) -> bool {
self.source.transports_delegation_context()
}
async fn capture_delegation_source(
&self,
request: &WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
self.ensure_active()?;
if self.scope.is_some() {
return Err(WorkdirError::Denied(
"scoped Workdir sessions cannot expose their provider source".into(),
));
}
self.source.capture_delegation_source(request).await
}
async fn delegate(
&self,
request: WorkdirDelegationRequest,
) -> Result<WorkdirDelegation, WorkdirError> {
let capabilities = self.validate_delegation_rules(&request.rules)?;
if !request
.rules
.iter()
.any(|rule| rule_allows_path(rule, &request.cwd, WorkdirDelegationPermission::Read))
{
return Err(WorkdirError::Denied(format!(
"delegated cwd `{}` is outside the delegated readable scope",
request.cwd
)));
}
let source = self.source.capture_delegation_source(&request).await?;
let validity = SessionValidity::child(self.validity.clone());
let id = self.next_lease_id.fetch_add(1, Ordering::Relaxed);
if request
.rules
.iter()
.any(|rule| rule.permission == WorkdirDelegationPermission::Write)
{
self.child_write_leases
.lock()
.expect("workdir delegation lease mutex poisoned")
.insert(
id,
ActiveWriteLease {
validity: Arc::downgrade(&validity),
rules: request.rules.clone(),
},
);
}
let child: WorkdirSessionHandle = Arc::new(DelegatingWorkdirSession {
source,
cwd: request.cwd,
scope: Some(request.rules),
capabilities,
validity: validity.clone(),
child_write_leases: Mutex::new(HashMap::new()),
next_lease_id: AtomicU64::new(1),
closes_source: false,
});
let scoped_session: WorkdirSessionHandle =
if capabilities == WorkdirSessionCapabilities::READ_ONLY {
Arc::new(ReadOnlyWorkdirSession::new(child))
} else {
child
};
Ok(WorkdirDelegation {
scoped_session,
capabilities,
validity,
})
}
async fn stat(&self, mut request: StatRequest) -> Result<StatResult, WorkdirError> {
let path = self.resolve_path(&request.path)?;
self.ensure_read(&path, WorkdirSessionCapability::Read)?;
if !self.source.transports_delegation_context() {
request.path = path;
}
self.source.stat(request).await
}
async fn read(&self, mut request: ReadRequest) -> Result<ReadResult, WorkdirError> {
let path = self.resolve_path(&request.path)?;
self.ensure_read(&path, WorkdirSessionCapability::Read)?;
if !self.source.transports_delegation_context() {
request.path = path;
}
self.source.read(request).await
}
async fn write(&self, mut request: WriteRequest) -> Result<WriteResult, WorkdirError> {
let path = self.resolve_path(&request.path)?;
self.ensure_write(&path, WorkdirSessionCapability::Write)?;
if !self.source.transports_delegation_context() {
request.path = path;
}
self.source.write(request).await
}
async fn edit(&self, mut request: EditRequest) -> Result<EditResult, WorkdirError> {
let path = self.resolve_path(&request.path)?;
self.ensure_write(&path, WorkdirSessionCapability::Edit)?;
if !self.source.transports_delegation_context() {
request.path = path;
}
self.source.edit(request).await
}
async fn list(&self, mut request: ListRequest) -> Result<ListResult, WorkdirError> {
let path = self.resolve_path(&request.path)?;
self.ensure_read(&path, WorkdirSessionCapability::Read)?;
if !self.source.transports_delegation_context() {
request.path = path;
}
self.source.list(request).await
}
async fn glob(&self, mut request: GlobRequest) -> Result<GlobResult, WorkdirError> {
let path = self.resolve_path(&request.path)?;
self.ensure_read(&path, WorkdirSessionCapability::Glob)?;
if !self.source.transports_delegation_context() {
request.path = path;
}
self.source.glob(request).await
}
async fn grep(&self, mut request: GrepRequest) -> Result<GrepResult, WorkdirError> {
let path = self.resolve_path(&request.path)?;
self.ensure_read(&path, WorkdirSessionCapability::Grep)?;
if !self.source.transports_delegation_context() {
request.path = path;
}
self.source.grep(request).await
}
async fn start_command(&self, request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
self.ensure_command(true)?;
self.source.start_command(request).await
}
async fn command_status(&self, handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
self.ensure_command(false)?;
self.source.command_status(handle).await
}
async fn command_output(
&self,
request: CommandOutputRequest,
) -> Result<CommandOutput, WorkdirError> {
self.ensure_command(false)?;
self.source.command_output(request).await
}
async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> {
self.ensure_command(false)?;
self.source.cancel_command(handle).await
}
async fn close(&self) -> Result<(), WorkdirError> {
self.validity.active.store(false, Ordering::Release);
if self.closes_source {
self.source.close().await
} else {
Ok(())
}
}
}
/// A fail-closed read-only view over an already scoped delegated session.
#[derive(Debug)]
pub struct ReadOnlyWorkdirSession {
inner: WorkdirSessionHandle,
}
impl ReadOnlyWorkdirSession {
pub fn new(inner: WorkdirSessionHandle) -> Self {
Self { inner }
}
}
#[async_trait]
impl WorkdirSession for ReadOnlyWorkdirSession {
fn workdir(&self) -> &Workdir {
self.inner.workdir()
}
fn capabilities(&self) -> WorkdirSessionCapabilities {
WorkdirSessionCapabilities::READ_ONLY
}
fn is_delegation_capable(&self) -> bool {
true
}
fn transports_delegation_context(&self) -> bool {
self.inner.transports_delegation_context()
}
async fn delegate(
&self,
request: WorkdirDelegationRequest,
) -> Result<WorkdirDelegation, WorkdirError> {
if request
.rules
.iter()
.any(|rule| rule.permission == WorkdirDelegationPermission::Write)
{
return Err(WorkdirError::Denied(
"read-only workdir session cannot delegate write access".into(),
));
}
self.inner.delegate(request).await
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
self.inner.stat(request).await
}
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
self.inner.read(request).await
}
async fn write(&self, _request: WriteRequest) -> Result<WriteResult, WorkdirError> {
Err(WorkdirError::Denied("read-only workdir session".into()))
}
async fn edit(&self, _request: EditRequest) -> Result<EditResult, WorkdirError> {
Err(WorkdirError::Denied("read-only workdir session".into()))
}
async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
self.inner.list(request).await
}
async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
self.inner.glob(request).await
}
async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
self.inner.grep(request).await
}
async fn start_command(&self, _request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
Err(WorkdirError::Denied("read-only workdir session".into()))
}
async fn command_status(&self, _handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
Err(WorkdirError::Denied("read-only workdir session".into()))
}
async fn command_output(
&self,
_request: CommandOutputRequest,
) -> Result<CommandOutput, WorkdirError> {
Err(WorkdirError::Denied("read-only workdir session".into()))
}
async fn cancel_command(&self, _handle: CommandHandle) -> Result<(), WorkdirError> {
Err(WorkdirError::Denied("read-only workdir session".into()))
}
async fn close(&self) -> Result<(), WorkdirError> {
self.inner.close().await
}
}
fn rule_allows_path(
rule: &WorkdirDelegationRule,
path: &FsPath,
required: WorkdirDelegationPermission,
) -> bool {
if required == WorkdirDelegationPermission::Write
&& rule.permission != WorkdirDelegationPermission::Write
{
return false;
}
path_in_rule(rule, path)
}
fn path_in_rule(rule: &WorkdirDelegationRule, path: &FsPath) -> bool {
let target = Path::new(rule.target.as_str());
let path = Path::new(path.as_str());
if path == target {
return true;
}
let Ok(suffix) = path.strip_prefix(target) else {
return false;
};
let depth = suffix.components().count();
rule.recursive || depth <= 1
}
fn rule_contains_rule(parent: &WorkdirDelegationRule, child: &WorkdirDelegationRule) -> bool {
if child.permission == WorkdirDelegationPermission::Write
&& parent.permission != WorkdirDelegationPermission::Write
{
return false;
}
if !path_in_rule(parent, &child.target) {
return false;
}
if parent.recursive {
return true;
}
!child.recursive && parent.target == child.target
}
#[cfg(test)]
mod tests {
use std::fs;
use manifest::{Permission, Scope, ScopeConfig, ScopeRule, SharedScope};
use tempfile::TempDir;
use super::*;
use crate::LocalWorkdirSession;
fn fs_path(path: &str) -> FsPath {
FsPath::new(path).unwrap()
}
fn session(root: &Path) -> WorkdirSessionHandle {
let scope = SharedScope::new(
Scope::from_config(&ScopeConfig {
allow: vec![ScopeRule {
target: root.to_path_buf(),
permission: Permission::Write,
recursive: true,
}],
deny: Vec::new(),
})
.unwrap(),
);
delegation_capable_session(Arc::new(LocalWorkdirSession::materialized_bound(
Workdir::new("delegation-test"),
root.to_path_buf(),
root.to_path_buf(),
scope,
WorkdirSessionCapabilities::ALL,
)))
}
fn request(path: &str, permission: WorkdirDelegationPermission) -> WorkdirDelegationRequest {
WorkdirDelegationRequest {
rules: vec![WorkdirDelegationRule {
target: fs_path(path),
permission,
recursive: true,
}],
cwd: fs_path(path),
}
}
fn read(path: &str) -> ReadRequest {
ReadRequest {
path: fs_path(path),
offset: 0,
limit: 20,
max_bytes: 1024,
}
}
fn write(path: &str, content: &str) -> WriteRequest {
WriteRequest {
path: fs_path(path),
content: content.as_bytes().to_vec(),
expected_hash: None,
}
}
#[test]
fn non_recursive_rule_covers_target_and_direct_children_only() {
let rule = WorkdirDelegationRule {
target: fs_path("docs"),
permission: WorkdirDelegationPermission::Read,
recursive: false,
};
assert!(path_in_rule(&rule, &fs_path("docs")));
assert!(path_in_rule(&rule, &fs_path("docs/readme.md")));
assert!(!path_in_rule(&rule, &fs_path("docs/guides/start.md")));
}
#[tokio::test]
async fn read_only_delegation_allows_prefix_and_denies_mutation() {
let root = TempDir::new().unwrap();
fs::create_dir_all(root.path().join("docs")).unwrap();
fs::create_dir_all(root.path().join("secret")).unwrap();
fs::write(root.path().join("docs/readme.md"), "visible").unwrap();
fs::write(root.path().join("secret/key"), "hidden").unwrap();
let parent = session(root.path());
let child = parent
.delegate(request("docs", WorkdirDelegationPermission::Read))
.await
.unwrap();
assert_eq!(child.capabilities, WorkdirSessionCapabilities::READ_ONLY);
assert_eq!(
child
.scoped_session
.read(read("readme.md"))
.await
.unwrap()
.bytes,
b"visible"
);
assert!(matches!(
child.scoped_session.write(write("new.md", "no")).await,
Err(WorkdirError::Denied(_))
));
assert!(
!child
.capabilities
.supports(WorkdirSessionCapability::Command)
);
}
#[cfg(unix)]
#[tokio::test]
async fn provider_scope_denies_read_through_symlink_outside_grant() {
use std::os::unix::fs::symlink;
let root = TempDir::new().unwrap();
fs::create_dir_all(root.path().join("granted")).unwrap();
fs::create_dir_all(root.path().join("secret")).unwrap();
fs::write(root.path().join("secret/key"), "hidden").unwrap();
symlink("../secret/key", root.path().join("granted/link")).unwrap();
let parent = session(root.path());
let child = parent
.delegate(request("granted", WorkdirDelegationPermission::Read))
.await
.unwrap();
let result = child.scoped_session.read(read("link")).await;
assert!(
result.is_err(),
"symlink read escaped provider scope: {result:?}"
);
}
#[cfg(unix)]
#[tokio::test]
async fn provider_scope_denies_write_through_symlink_outside_grant() {
use std::os::unix::fs::symlink;
let root = TempDir::new().unwrap();
fs::create_dir_all(root.path().join("granted")).unwrap();
fs::create_dir_all(root.path().join("secret")).unwrap();
symlink("../secret", root.path().join("granted/outside")).unwrap();
let parent = session(root.path());
let child = parent
.delegate(request("granted", WorkdirDelegationPermission::Write))
.await
.unwrap();
let result = child
.scoped_session
.write(write("outside/new", "forbidden"))
.await;
assert!(
result.is_err(),
"symlink write escaped provider scope: {result:?}"
);
assert!(!root.path().join("secret/new").exists());
}
#[cfg(unix)]
#[tokio::test]
async fn write_delegation_rejects_symlink_target_before_lease() {
use std::os::unix::fs::symlink;
let root = TempDir::new().unwrap();
fs::create_dir_all(root.path().join("granted")).unwrap();
fs::create_dir_all(root.path().join("secret")).unwrap();
symlink("../secret", root.path().join("granted/outside")).unwrap();
let parent = session(root.path());
assert!(matches!(
parent
.delegate(request(
"granted/outside",
WorkdirDelegationPermission::Write
))
.await,
Err(WorkdirError::Denied(_))
));
parent
.write(write("secret/parent", "still-authoritative"))
.await
.unwrap();
}
#[tokio::test]
async fn write_lease_blocks_parent_region_until_release() {
let root = TempDir::new().unwrap();
fs::create_dir_all(root.path().join("leased")).unwrap();
fs::create_dir_all(root.path().join("other")).unwrap();
let parent = session(root.path());
let child = parent
.delegate(request("leased", WorkdirDelegationPermission::Write))
.await
.unwrap();
assert!(matches!(
parent.write(write("leased/file", "parent")).await,
Err(WorkdirError::Denied(_))
));
parent.write(write("other/file", "parent")).await.unwrap();
child
.scoped_session
.write(write("file", "child"))
.await
.unwrap();
child.release();
parent
.write(write("leased/parent", "parent"))
.await
.unwrap();
assert!(matches!(
child.scoped_session.read(read("file")).await,
Err(WorkdirError::SessionClosed)
));
}
#[tokio::test]
async fn nested_delegation_is_attenuated_and_parent_revocation_cascades() {
let root = TempDir::new().unwrap();
fs::create_dir_all(root.path().join("docs/sub")).unwrap();
fs::create_dir_all(root.path().join("docs/peer")).unwrap();
fs::write(root.path().join("docs/sub/a"), "a").unwrap();
fs::write(root.path().join("docs/peer/b"), "b").unwrap();
let root_session = session(root.path());
let child = root_session
.delegate(request("docs", WorkdirDelegationPermission::Read))
.await
.unwrap();
let nested = child
.scoped_session
.delegate(request("docs/sub", WorkdirDelegationPermission::Read))
.await
.unwrap();
nested.scoped_session.read(read("a")).await.unwrap();
assert!(
child
.scoped_session
.delegate(request("other", WorkdirDelegationPermission::Read))
.await
.is_err()
);
assert!(
child
.scoped_session
.delegate(request("docs/sub", WorkdirDelegationPermission::Write))
.await
.is_err()
);
child.release();
assert!(matches!(
nested.scoped_session.read(read("a")).await,
Err(WorkdirError::SessionClosed)
));
}
#[tokio::test]
async fn applied_chain_cannot_replace_outer_provider_attenuation() {
let root = TempDir::new().unwrap();
fs::create_dir_all(root.path().join("outer")).unwrap();
fs::create_dir_all(root.path().join("outside")).unwrap();
let result = apply_delegation_chain(
Arc::new(LocalWorkdirSession::materialized_bound(
Workdir::new("delegation-chain-test"),
root.path().to_path_buf(),
root.path().to_path_buf(),
SharedScope::new(
Scope::from_config(&ScopeConfig {
allow: vec![ScopeRule {
target: root.path().to_path_buf(),
permission: Permission::Write,
recursive: true,
}],
deny: Vec::new(),
})
.unwrap(),
),
WorkdirSessionCapabilities::ALL,
)),
[
request("outer", WorkdirDelegationPermission::Read),
request("outside", WorkdirDelegationPermission::Read),
],
)
.await;
assert!(matches!(result, Err(WorkdirError::Denied(_))));
}
#[tokio::test]
async fn closing_parent_invalidates_delegated_sessions() {
let root = TempDir::new().unwrap();
fs::create_dir_all(root.path().join("docs")).unwrap();
fs::write(root.path().join("docs/a"), "a").unwrap();
let parent = session(root.path());
let child = parent
.delegate(request("docs", WorkdirDelegationPermission::Read))
.await
.unwrap();
parent.close().await.unwrap();
assert!(matches!(
child.scoped_session.read(read("a")).await,
Err(WorkdirError::SessionClosed)
));
}
}
+51 -2
View File
@@ -68,6 +68,15 @@ pub enum WorkdirSessionOperation {
CommandCancel(CommandHandle),
}
/// Wire envelope for an operation and its optional provider-enforced child scope.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkdirSessionOperationRequest {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub delegations: Vec<crate::WorkdirDelegationRequest>,
pub operation: WorkdirSessionOperation,
}
/// Typed result paired with [`WorkdirSessionOperation`].
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "operation", content = "result", rename_all = "snake_case")]
@@ -120,7 +129,10 @@ impl WorkdirTransportError {
WorkdirError::UnknownCommand(_) => {
(Code::UnknownCommand, "Workdir command was not found")
}
WorkdirError::Unavailable(_) => (Code::Unavailable, "Workdir session is unavailable"),
WorkdirError::Unavailable(_) | WorkdirError::SessionClosed => {
(Code::Unavailable, "Workdir session is unavailable")
}
WorkdirError::Denied(_) => (Code::InvalidRequest, "Workdir operation was denied"),
WorkdirError::Transport(_) => (Code::Internal, "Workdir transport failed"),
WorkdirError::InvalidPath(_)
| WorkdirError::RelativePath(_)
@@ -169,7 +181,7 @@ mod client {
use reqwest::{Client, StatusCode, Url};
use super::*;
use crate::{Workdir, WorkdirSession};
use crate::{Workdir, WorkdirSession, WorkdirSessionHandle};
/// Provides a fresh bearer token for each Runtime request. Backend
/// implementations can mint short-lived capability tokens without making a
@@ -204,6 +216,7 @@ mod client {
workdir: Workdir,
session_id: WorkdirSessionId,
capabilities: WorkdirSessionCapabilities,
delegations: Vec<crate::WorkdirDelegationRequest>,
closed: AtomicBool,
}
@@ -256,6 +269,7 @@ mod client {
workdir: Workdir::new(opened.workdir_id.as_str()),
session_id: opened.session_id,
capabilities: opened.capabilities,
delegations: Vec::new(),
closed: AtomicBool::new(false),
})
}
@@ -282,6 +296,10 @@ mod client {
"operations",
],
)?;
let operation = WorkdirSessionOperationRequest {
delegations: self.delegations.clone(),
operation,
};
let response = self
.client
.post(url)
@@ -310,6 +328,37 @@ mod client {
self.capabilities
}
fn transports_delegation_context(&self) -> bool {
true
}
async fn capture_delegation_source(
&self,
request: &crate::WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
if self.closed.load(Ordering::Acquire) {
return Err(WorkdirError::SessionClosed);
}
let mut delegations = self.delegations.clone();
delegations.push(request.clone());
let candidate = Arc::new(Self {
client: self.client.clone(),
base_url: self.base_url.clone(),
authorization: self.authorization.clone(),
workdir: self.workdir.clone(),
session_id: self.session_id.clone(),
capabilities: self.capabilities,
delegations,
closed: AtomicBool::new(false),
});
candidate
.stat(StatRequest {
path: fs_operation::FsPath::new("").expect("empty Workdir path is valid"),
})
.await?;
Ok(candidate)
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
match self.operate(WorkdirSessionOperation::Stat(request)).await? {
WorkdirSessionOperationResult::Stat(result) => Ok(result),
+45 -100
View File
@@ -5,6 +5,7 @@
//! bound to one Worker. Tools consume sessions; they do not own Workdir
//! materialization or cleanup.
mod delegation;
pub mod http;
mod local;
mod operation;
@@ -12,11 +13,15 @@ pub mod workspace;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
pub use delegation::{
AppliedWorkdirDelegation, ReadOnlyWorkdirSession, WorkdirDelegation,
WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule,
apply_delegation_chain, delegation_capable_session,
};
pub use fs_operation::{
ContentHash, EditRequest, EditResult, EntryKind, FsPath as WorkdirPath, GlobRequest,
GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult,
@@ -140,6 +145,39 @@ pub trait WorkdirSession: std::fmt::Debug + Send + Sync {
fn workdir(&self) -> &Workdir;
fn capabilities(&self) -> WorkdirSessionCapabilities;
fn is_delegation_capable(&self) -> bool {
false
}
/// Whether this session transports the delegation chain to another
/// provider boundary that will apply logical cwd/path resolution there.
fn transports_delegation_context(&self) -> bool {
false
}
/// Capture a provider-specific source for a delegated child session.
/// Remote providers use this boundary to pin attachment identity without
/// exposing transport handles or host paths.
async fn capture_delegation_source(
&self,
_request: &WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
Err(WorkdirError::Denied(
"workdir provider does not support delegated sessions".into(),
))
}
/// Attenuate this session into a revocable child lease. Only sessions
/// created with [`delegation_capable_session`] implement this operation.
async fn delegate(
&self,
_request: WorkdirDelegationRequest,
) -> Result<WorkdirDelegation, WorkdirError> {
Err(WorkdirError::Denied(
"workdir session is not delegation-capable".into(),
))
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError>;
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError>;
async fn write(&self, request: WriteRequest) -> Result<WriteResult, WorkdirError>;
@@ -160,107 +198,14 @@ pub trait WorkdirSession: std::fmt::Debug + Send + Sync {
pub type WorkdirSessionHandle = Arc<dyn WorkdirSession>;
/// Ephemeral least-authority view over an existing Workdir session.
///
/// The wrapper exposes only stat/read/list/glob/grep and never forwards write,
/// edit, command, or close authority to the underlying Worker session. Closing
/// the wrapper is terminal for the view but deliberately leaves the owner's
/// source session open.
#[derive(Debug)]
pub struct ReadOnlyWorkdirSession {
source: WorkdirSessionHandle,
closed: AtomicBool,
}
impl ReadOnlyWorkdirSession {
pub fn new(source: WorkdirSessionHandle) -> Self {
Self {
source,
closed: AtomicBool::new(false),
}
}
fn ensure_open(&self) -> Result<(), WorkdirError> {
if self.closed.load(Ordering::Acquire) {
Err(WorkdirError::Unavailable(
"read-only Workdir session is closed".to_string(),
))
} else {
Ok(())
}
}
}
#[async_trait]
impl WorkdirSession for ReadOnlyWorkdirSession {
fn workdir(&self) -> &Workdir {
self.source.workdir()
}
fn capabilities(&self) -> WorkdirSessionCapabilities {
WorkdirSessionCapabilities::READ_ONLY
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
self.ensure_open()?;
self.source.stat(request).await
}
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
self.ensure_open()?;
self.source.read(request).await
}
async fn write(&self, _request: WriteRequest) -> Result<WriteResult, WorkdirError> {
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Write))
}
async fn edit(&self, _request: EditRequest) -> Result<EditResult, WorkdirError> {
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Edit))
}
async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
self.ensure_open()?;
self.source.list(request).await
}
async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
self.ensure_open()?;
self.source.glob(request).await
}
async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
self.ensure_open()?;
self.source.grep(request).await
}
async fn start_command(&self, _request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
}
async fn command_status(&self, _handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
}
async fn command_output(
&self,
_request: CommandOutputRequest,
) -> Result<CommandOutput, WorkdirError> {
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
}
async fn cancel_command(&self, _handle: CommandHandle) -> Result<(), WorkdirError> {
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
}
async fn close(&self) -> Result<(), WorkdirError> {
self.closed.store(true, Ordering::Release);
Ok(())
}
}
#[derive(Debug, thiserror::Error)]
pub enum WorkdirError {
#[error("Workdir operation denied: {0}")]
Denied(String),
#[error("Workdir session is closed")]
SessionClosed,
#[error("Workdir session does not support {0:?}")]
Unsupported(WorkdirSessionCapability),
+67 -3
View File
@@ -19,7 +19,7 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
use async_trait::async_trait;
use manifest::{Scope, SharedScope};
use manifest::{Permission, Scope, ScopeConfig, ScopeRule, SharedScope};
use sha2::{Digest, Sha256};
use tokio::process::Command;
use tokio::sync::{Mutex, Notify};
@@ -28,8 +28,9 @@ use tokio::task::JoinHandle;
use crate::{
CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest,
EditResult, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult,
ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirError, WorkdirPath,
WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability, WriteRequest,
ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirDelegationPermission,
WorkdirDelegationRequest, WorkdirError, WorkdirPath, WorkdirSession,
WorkdirSessionCapabilities, WorkdirSessionCapability, WorkdirSessionHandle, WriteRequest,
WriteResult,
};
#[cfg(test)]
@@ -371,6 +372,69 @@ impl WorkdirSession for LocalWorkdirSession {
self.inner.capabilities
}
async fn capture_delegation_source(
&self,
request: &WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
let host_rules = request
.rules
.iter()
.map(|rule| ScopeRule {
target: self.inner.root.join(rule.target.as_str()),
permission: match rule.permission {
WorkdirDelegationPermission::Read => Permission::Read,
WorkdirDelegationPermission::Write => Permission::Write,
},
recursive: rule.recursive,
})
.collect::<Vec<_>>();
for (logical, host) in request.rules.iter().zip(&host_rules) {
if logical.permission == WorkdirDelegationPermission::Write {
let resolved = Scope::resolved_target(host)
.map_err(|error| WorkdirError::Denied(error.to_string()))?;
if resolved != host.target {
return Err(WorkdirError::Denied(format!(
"write delegation target `{}` traverses a symlink",
logical.target
)));
}
}
}
let parent_scope = self.inner.scope.snapshot();
for rule in &host_rules {
if !parent_scope
.allows_rule(rule)
.map_err(|error| WorkdirError::Denied(error.to_string()))?
{
return Err(WorkdirError::Denied(format!(
"delegated provider scope `{}` exceeds the parent session",
rule.target.display()
)));
}
}
let child_scope = Scope::from_config(&ScopeConfig {
allow: host_rules,
deny: Vec::new(),
})
.map_err(|error| WorkdirError::Denied(error.to_string()))?;
let child_cwd = self.inner.root.join(request.cwd.as_str());
if !child_scope.is_readable(&child_cwd)
|| !std::fs::metadata(&child_cwd).is_ok_and(|metadata| metadata.is_dir())
{
return Err(WorkdirError::Denied(format!(
"delegated cwd `{}` is not a readable Workdir directory",
request.cwd
)));
}
Ok(Arc::new(LocalWorkdirSession::materialized_bound(
self.inner.workdir.clone(),
self.inner.root.clone(),
self.inner.root.clone(),
SharedScope::new(child_scope),
self.inner.capabilities,
)))
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
self.ensure_capability(WorkdirSessionCapability::Read)?;
let logical = request.path.clone();
+16
View File
@@ -314,3 +314,19 @@ mod tests {
assert_eq!(decoded, detail);
}
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkspaceWorkdirSessionOperationRequest {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub expected_session_fence: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub delegations: Vec<crate::WorkdirDelegationRequest>,
pub operation: crate::http::WorkdirSessionOperation,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkspaceWorkdirSessionFence {
pub value: String,
}
+88 -8
View File
@@ -59,8 +59,8 @@ use workdir::{
CommandOutput, CommandStatus, WorkdirSessionHandle,
http::{
OpenWorkdirSessionRequest, OpenWorkdirSessionResponse, WorkdirSessionId,
WorkdirSessionOperation, WorkdirSessionOperationResult, WorkdirTransportError,
WorkdirTransportErrorCode,
WorkdirSessionOperation, WorkdirSessionOperationRequest, WorkdirSessionOperationResult,
WorkdirTransportError, WorkdirTransportErrorCode,
},
};
@@ -596,11 +596,11 @@ async fn run_workdir_session_operation(
State(state): State<RuntimeHttpState>,
Path(session_id): Path<String>,
auth: Option<Extension<RuntimeAuthContext>>,
body: Result<Json<WorkdirSessionOperation>, JsonRejection>,
body: Result<Json<WorkdirSessionOperationRequest>, JsonRejection>,
) -> Result<Json<WorkdirSessionOperationResult>, RuntimeHttpWorkdirError> {
let Json(operation) = body.map_err(|_| RuntimeHttpWorkdirError::invalid_request())?;
let Json(request) = body.map_err(|_| RuntimeHttpWorkdirError::invalid_request())?;
let owner = required_workdir_owner(auth)?;
let session = {
let source = {
let sessions = state
.workdir_sessions
.lock()
@@ -611,6 +611,9 @@ async fn run_workdir_session_operation(
.ok_or_else(RuntimeHttpWorkdirError::not_found)?;
record.session.clone()
};
let applied = workdir::apply_delegation_chain(source, request.delegations).await?;
let session = applied.scoped_session.as_ref();
let operation = request.operation;
let result = match operation {
WorkdirSessionOperation::Stat(request) => {
@@ -1836,7 +1839,8 @@ mod tests {
use manifest::{Scope, SharedScope};
use tower::ServiceExt;
use workdir::{
LocalWorkdirSession, StatRequest, Workdir, WorkdirPath, WorkdirSessionCapabilities,
LocalWorkdirSession, ReadRequest, StatRequest, Workdir, WorkdirPath,
WorkdirSessionCapabilities,
};
fn test_bundle(profile: ProfileSelector) -> ConfigBundle {
@@ -2224,6 +2228,16 @@ mod tests {
async fn workdir_session_operations_enforce_owner_and_close_terminally() {
let temp = tempfile::tempdir().expect("tempdir");
std::fs::write(temp.path().join("hello.txt"), "hello").expect("write fixture");
#[cfg(unix)]
{
use std::os::unix::fs::symlink;
std::fs::create_dir(temp.path().join("granted")).expect("granted directory");
std::fs::write(temp.path().join("granted/visible"), "visible")
.expect("visible fixture");
std::fs::create_dir(temp.path().join("secret")).expect("secret directory");
std::fs::write(temp.path().join("secret/key"), "hidden").expect("secret fixture");
symlink("../secret/key", temp.path().join("granted/link")).expect("symlink fixture");
}
let scope = SharedScope::new(Scope::writable(temp.path()).expect("scope"));
let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound(
Workdir::new("wd-1"),
@@ -2256,9 +2270,12 @@ mod tests {
token_id: "token-a".to_string(),
expires_at: u64::MAX,
};
let operation = WorkdirSessionOperation::Stat(StatRequest {
let operation = WorkdirSessionOperationRequest {
delegations: Vec::new(),
operation: WorkdirSessionOperation::Stat(StatRequest {
path: WorkdirPath::new("hello.txt").expect("logical path"),
});
}),
};
let Json(result) = run_workdir_session_operation(
State(state.clone()),
@@ -2270,6 +2287,69 @@ mod tests {
.expect("owned operation");
assert!(matches!(result, WorkdirSessionOperationResult::Stat(_)));
#[cfg(unix)]
{
let delegated_visible = WorkdirSessionOperationRequest {
delegations: vec![workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: WorkdirPath::new("granted").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: true,
}],
cwd: WorkdirPath::new("granted").unwrap(),
}],
operation: WorkdirSessionOperation::Read(ReadRequest {
path: WorkdirPath::new("visible").unwrap(),
offset: 0,
limit: 20,
max_bytes: 1024,
}),
};
let visible = run_workdir_session_operation(
State(state.clone()),
Path("session-1".to_string()),
Some(Extension(auth.clone())),
Ok(Json(delegated_visible)),
)
.await
.expect("non-root delegated cwd should resolve once")
.0;
assert!(matches!(
visible,
WorkdirSessionOperationResult::Read(result) if result.bytes == b"visible"
));
let delegated_read = WorkdirSessionOperationRequest {
delegations: vec![workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: WorkdirPath::new("granted").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: true,
}],
cwd: WorkdirPath::new("granted").unwrap(),
}],
operation: WorkdirSessionOperation::Read(ReadRequest {
path: WorkdirPath::new("link").unwrap(),
offset: 0,
limit: 20,
max_bytes: 1024,
}),
};
let error = run_workdir_session_operation(
State(state.clone()),
Path("session-1".to_string()),
Some(Extension(auth.clone())),
Ok(Json(delegated_read)),
)
.await
.expect_err("provider must reject delegated symlink escape");
assert_ne!(error.status, StatusCode::OK);
assert_eq!(
std::fs::read_to_string(temp.path().join("secret/key")).unwrap(),
"hidden"
);
}
let wrong_owner = RuntimeAuthContext {
workspace_id: "workspace-b".to_string(),
..auth.clone()
+2
View File
@@ -1435,6 +1435,7 @@ impl Runtime {
},
status: protocol::WorkerStatus::Idle,
in_flight: protocol::InFlightSnapshot { blocks: Vec::new() },
internal_workers: Vec::new(),
})
}
@@ -3770,6 +3771,7 @@ mod tests {
},
status: protocol::WorkerStatus::Running,
in_flight: protocol::InFlightSnapshot { blocks: Vec::new() },
internal_workers: Vec::new(),
},
);
+1
View File
@@ -33,6 +33,7 @@ config-source = { path = "../config-source" }
include_dir = "0.7.4"
fs4 = { workspace = true, features = ["sync"] }
flow = { path = "../flow" }
fs-operation = { workspace = true }
libc = { workspace = true }
schemars = { workspace = true }
ticket = { workspace = true }
+24 -30
View File
@@ -48,6 +48,7 @@ pub struct WorkerHandle {
/// it on every new connection (Event::Snapshot) and forwards
/// subsequent commits (Event::Entry) on the receiver.
pub sink: SegmentLogSink,
spawned_registry: Arc<SpawnedWorkerRegistry>,
}
impl WorkerHandle {
@@ -84,6 +85,7 @@ impl WorkerHandle {
greeting: self.shared_state.greeting.clone(),
status: self.shared_state.get_status(),
in_flight,
internal_workers: self.spawned_registry.internal_worker_snapshots(),
};
(event, entry_rx)
}
@@ -413,6 +415,7 @@ impl WorkerController {
wire_event_bridges_on_engine(&mut worker, &event_tx, &alerter, &in_flight);
// === 3. Tool registration (builtin / memory / spawn-orchestration) ===
spawned_registry.attach_parent_protocol(event_tx.clone(), worker.session_id().to_string());
let fs_for_view = register_worker_tools(
&mut worker,
bash_output_dir,
@@ -460,6 +463,7 @@ impl WorkerController {
alerter: alerter.clone(),
in_flight: in_flight.clone(),
sink: worker.sink(),
spawned_registry: spawned_registry.clone(),
};
let socket_server = match transport {
@@ -502,7 +506,7 @@ impl WorkerController {
/// per-item history commit callback so every assistant / tool item
/// landing in `worker.history` becomes a singular `LogEntry::AssistantItem`
/// / `ToolResult` commit through the sync writer.
fn wire_event_bridges_on_engine<C, St>(
pub(crate) fn wire_event_bridges_on_engine<C, St>(
worker: &mut Worker<C, St>,
event_tx: &broadcast::Sender<Event>,
alerter: &Alerter,
@@ -681,18 +685,20 @@ where
{
// Worker-immutable snapshots taken before the mutable worker borrow
// below so the worker borrow doesn't conflict with reads on `worker`.
let scope_handle = worker.scope().clone();
let feature_config = worker.manifest().feature.clone();
if feature_config.manage_workdir.enabled {
if let Some(existing) = worker.workdir_session().cloned() {
existing.close().await.map_err(std::io::Error::other)?;
}
if feature_config.manage_workdir.enabled && worker.workdir_session().is_none() {
let workspace_client = worker.workspace_client_handle();
worker.bind_workdir_session(Some(
worker.bind_workdir_session(Some(workdir::delegation_capable_session(
crate::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession::handle(
workspace_client,
),
));
)));
}
if feature_config.sub_worker.enabled
&& let Some(existing) = worker.workdir_session().cloned()
&& !existing.is_delegation_capable()
{
worker.bind_workdir_session(Some(workdir::delegation_capable_session(existing)));
}
let worker_workdir = worker.workdir_session().cloned();
let local_filesystem = worker.local_working_directory().cloned();
@@ -844,6 +850,7 @@ where
}
let host_worker_observation_provider = worker.worker_observation_provider();
let source_workdir_session = worker.workdir_session().cloned();
{
let workspace_client = worker.workspace_client_handle();
let engine = worker.engine_mut();
@@ -902,37 +909,23 @@ where
Arc<dyn crate::feature::builtin::worker_observation::WorkerObservationProvider>,
> = Vec::new();
// Worker-orchestration tools (SubWorkerSpawn + three control tools) share
// the Worker-scoped `SpawnedWorkerRegistry` (also consumed by the main
// loop's `WorkerEvent` handler). Expose them only behind the explicit
// profile feature and require delegation authority up front so enabling
// the surface cannot imply broad child scope by accident.
// Worker-orchestration tools derive child filesystem authority from the
// active provider-backed Workdir session. The tool remains registered
// without one so invocation fails deterministically until the parent
// attaches a Workdir.
if feature_config.sub_worker.enabled {
let spawner_cwd = local_filesystem
.as_ref()
.map(|local| local.cwd.clone())
.ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"worker spawn tools require local Worker filesystem authority",
)
})?;
let spawner_workspace_root = local_workspace_root.clone().ok_or_else(|| {
std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"worker spawn tools require local Worker filesystem authority",
)
})?;
let spawner_workspace_root = local_workspace_root
.clone()
.unwrap_or_else(|| PathBuf::from("/"));
engine.register_tool(sub_worker_spawn_tool(
spawner_name.clone(),
spawner_workspace_context,
parent_notifications,
runtime_base.clone(),
spawner_workspace_root,
spawner_cwd.clone(),
source_workdir_session,
spawned_registry.clone(),
spawner_manifest,
scope_handle,
prompts,
));
observation_providers.push(Arc::new(
@@ -1888,6 +1881,7 @@ mod tests {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await
.ok()?;
+6
View File
@@ -1494,6 +1494,7 @@ mod tests {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await
.unwrap();
@@ -1526,6 +1527,7 @@ mod tests {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await
.unwrap();
@@ -1614,6 +1616,7 @@ mod tests {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await
.unwrap();
@@ -1637,6 +1640,7 @@ mod tests {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await
.unwrap();
@@ -1738,6 +1742,7 @@ mod tests {
},
status: WorkerStatus::Paused,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await
.unwrap();
@@ -1787,6 +1792,7 @@ mod tests {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
})
.await;
});
@@ -16,7 +16,8 @@ use serde_json::json;
use workdir::http::{WorkdirSessionOperation, WorkdirSessionOperationResult};
use workdir::workspace::{
WorkingDirectoryDetailResponse as WorkdirDetailResponse,
WorkingDirectoryListResponse as WorkdirListResponse,
WorkingDirectoryListResponse as WorkdirListResponse, WorkspaceWorkdirSessionFence,
WorkspaceWorkdirSessionOperationRequest,
};
use workdir::{
CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest,
@@ -153,6 +154,8 @@ struct WorkspaceHttpWorkdirBackend {
pub struct WorkspaceAttachedWorkdirSession {
client: Arc<dyn WorkspaceClient>,
workdir: Workdir,
expected_session_fence: Option<String>,
delegations: Vec<workdir::WorkdirDelegationRequest>,
}
impl WorkspaceAttachedWorkdirSession {
@@ -160,6 +163,8 @@ impl WorkspaceAttachedWorkdirSession {
Arc::new(Self {
client,
workdir: Workdir::new("workspace-attachment"),
expected_session_fence: None,
delegations: Vec::new(),
})
}
@@ -176,7 +181,12 @@ impl WorkspaceAttachedWorkdirSession {
"/api/w/{}/workers/self/workdir-session/operations",
encode_path_segment(workspace_id)
),
serde_json::to_string(&operation).map_err(|error| {
serde_json::to_string(&WorkspaceWorkdirSessionOperationRequest {
expected_session_fence: self.expected_session_fence.clone(),
delegations: self.delegations.clone(),
operation,
})
.map_err(|error| {
WorkdirError::Transport(format!(
"failed to encode Workspace Workdir operation: {error}"
))
@@ -224,6 +234,59 @@ impl WorkdirSession for WorkspaceAttachedWorkdirSession {
WorkdirSessionCapabilities::ALL
}
fn transports_delegation_context(&self) -> bool {
true
}
async fn capture_delegation_source(
&self,
request: &workdir::WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
let expected_session_fence = if let Some(fence) = &self.expected_session_fence {
fence.clone()
} else {
let workspace_id = self.client.workspace_id().ok_or_else(|| {
WorkdirError::Unavailable("Workspace identity is unavailable".to_string())
})?;
let response = self
.client
.execute(WorkspaceRequest {
method: WorkspaceRequestMethod::Get,
path: format!(
"/api/w/{}/workers/self/workdir-session/fence",
encode_path_segment(workspace_id)
),
body: None,
})
.map_err(|error| {
WorkdirError::Unavailable(format!(
"failed to capture Workdir attachment fence: {error}"
))
})?;
let fence: WorkspaceWorkdirSessionFence = serde_json::from_str(&response.body)
.map_err(|error| {
WorkdirError::Unavailable(format!(
"invalid Workdir attachment fence response: {error}"
))
})?;
fence.value
};
let mut delegations = self.delegations.clone();
delegations.push(request.clone());
let candidate = Arc::new(Self {
client: self.client.clone(),
workdir: self.workdir.clone(),
expected_session_fence: Some(expected_session_fence),
delegations,
});
candidate
.stat(StatRequest {
path: workdir::WorkdirPath::new("").expect("empty Workdir path is valid"),
})
.await?;
Ok(candidate)
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
match self.operate(WorkdirSessionOperation::Stat(request))? {
WorkdirSessionOperationResult::Stat(result) => Ok(result),
@@ -1002,11 +1065,159 @@ mod tests {
);
let body: serde_json::Value =
serde_json::from_str(requests[0].body.as_deref().unwrap()).unwrap();
assert_eq!(body["operation"], "stat");
assert_eq!(body["operation"]["operation"], "stat");
assert!(body.get("expected_session_fence").is_none());
assert!(body.get("runtime_id").is_none());
assert!(body.get("session_id").is_none());
}
#[tokio::test]
async fn delegated_attached_session_carries_captured_fence_on_operations() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![
response(json!({"value": "attachment-fence"})),
response(json!({
"operation": "stat",
"result": {"path": "", "kind": "directory", "size": 0}
})),
response(json!({
"operation": "stat",
"result": {"path": "visible.txt", "kind": "file", "size": 8}
})),
]));
let parent = workdir::delegation_capable_session(WorkspaceAttachedWorkdirSession::handle(
client.clone(),
));
let delegation = parent
.delegate(workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: workdir::WorkdirPath::new("").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: false,
}],
cwd: workdir::WorkdirPath::new("").unwrap(),
})
.await
.unwrap();
delegation
.scoped_session
.stat(StatRequest {
path: workdir::WorkdirPath::new("visible.txt").unwrap(),
})
.await
.unwrap();
let requests = client.requests();
assert_eq!(requests.len(), 3);
assert_eq!(
requests[0].path,
"/api/w/workspace%2Ftest/workers/self/workdir-session/fence"
);
let body: serde_json::Value =
serde_json::from_str(requests[2].body.as_deref().unwrap()).unwrap();
assert_eq!(body["expected_session_fence"], "attachment-fence");
assert_eq!(body["operation"]["operation"], "stat");
assert_eq!(body["delegations"][0]["rules"][0]["target"], "");
}
#[tokio::test]
async fn attached_provider_rejection_happens_before_delegation_is_returned() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![
response(json!({"value": "attachment-fence"})),
response(json!({"error": "provider rejected delegated write target"})),
]));
let parent = workdir::delegation_capable_session(WorkspaceAttachedWorkdirSession::handle(
client.clone(),
));
let result = parent
.delegate(workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: workdir::WorkdirPath::new("linked-target").unwrap(),
permission: workdir::WorkdirDelegationPermission::Write,
recursive: true,
}],
cwd: workdir::WorkdirPath::new("linked-target").unwrap(),
})
.await;
assert!(result.is_err(), "provider rejection must fail before lease");
let requests = client.requests();
assert_eq!(requests.len(), 2);
let validation: serde_json::Value =
serde_json::from_str(requests[1].body.as_deref().unwrap()).unwrap();
assert_eq!(validation["operation"]["operation"], "stat");
assert_eq!(validation["delegations"].as_array().unwrap().len(), 1);
}
#[tokio::test]
async fn nested_attached_session_preserves_full_delegation_chain() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![
response(json!({"value": "attachment-fence"})),
response(json!({
"operation": "stat",
"result": {"path": "", "kind": "directory", "size": 0}
})),
response(json!({
"operation": "stat",
"result": {"path": "nested", "kind": "directory", "size": 0}
})),
response(json!({
"operation": "stat",
"result": {"path": "nested/file", "kind": "file", "size": 1}
})),
]));
let parent = workdir::delegation_capable_session(WorkspaceAttachedWorkdirSession::handle(
client.clone(),
));
let outer = parent
.delegate(workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: workdir::WorkdirPath::new("").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: true,
}],
cwd: workdir::WorkdirPath::new("").unwrap(),
})
.await
.unwrap();
let nested = outer
.scoped_session
.delegate(workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: workdir::WorkdirPath::new("nested").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: true,
}],
cwd: workdir::WorkdirPath::new("nested").unwrap(),
})
.await
.unwrap();
nested
.scoped_session
.stat(StatRequest {
path: workdir::WorkdirPath::new("file").unwrap(),
})
.await
.unwrap();
let requests = client.requests();
assert_eq!(requests.len(), 4);
let outer_validation: serde_json::Value =
serde_json::from_str(requests[1].body.as_deref().unwrap()).unwrap();
let nested_validation: serde_json::Value =
serde_json::from_str(requests[2].body.as_deref().unwrap()).unwrap();
assert_eq!(outer_validation["delegations"].as_array().unwrap().len(), 1);
assert_eq!(
nested_validation["delegations"].as_array().unwrap().len(),
2
);
let body: serde_json::Value =
serde_json::from_str(requests[3].body.as_deref().unwrap()).unwrap();
assert_eq!(body["delegations"].as_array().unwrap().len(), 2);
assert_eq!(body["delegations"][0]["rules"][0]["target"], "");
assert_eq!(body["delegations"][1]["rules"][0]["target"], "nested");
assert_eq!(body["operation"]["request"]["path"], "file");
}
#[test]
fn invalid_or_extra_inputs_are_rejected_before_workspace_request() {
let client = Arc::new(RecordingWorkspaceClient::new(Vec::new()));
+203 -6
View File
@@ -12,10 +12,18 @@ use std::sync::{Arc, Mutex};
use llm_engine::timeline::event::UsageEvent;
use llm_engine::{Engine, llm_client::LlmClient};
use manifest::{Scope, WorkerManifest};
use protocol::{Event, InFlightSnapshot, WorkerStatus};
use session_store::{LogEntry, SegmentId, SessionId, Store, StoreError, TraceEntry};
use tokio::sync::broadcast;
use uuid::Uuid;
use crate::controller::wire_event_bridges_on_engine;
use crate::feature::FeatureRegistryBuilder;
use crate::in_flight::{InFlightEvents, snapshot_from_guard};
use crate::ipc::alerter::Alerter;
use crate::ipc::protocol_session::live_log_entry_event;
use crate::segment_log_sink::SegmentLogSink;
use crate::spawn::registry::SpawnedWorkerRegistry;
use crate::worker::{
Worker, WorkerError, WorkerFilesystemAuthority, WorkerRunResult, WorkerWorkspaceContext,
};
@@ -195,6 +203,20 @@ where
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InternalWorkerVisibility {
/// Output may be projected only through the owning parent's protocol stream.
ParentClient,
/// Backend-owned helper output remains private to the service authority.
ServicePrivate,
}
impl Default for InternalWorkerVisibility {
fn default() -> Self {
Self::ServicePrivate
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum InternalWorkerSessionStatus {
Idle,
@@ -246,8 +268,18 @@ enum InternalWorkerSessionCommand {
/// Parent-owned handle for a long-lived Internal Worker session.
///
/// The handle exposes only typed turn, history, status, and stop operations. The underlying Worker,
/// Engine, ephemeral Store, and cancellation sender remain inside the actor task.
/// The handle exposes typed turn, history, status, presentation snapshot, event subscription, and
/// stop operations. The underlying Worker, Engine, and cancellation sender remain inside the actor
/// task; protocol access is consumed only by the owning parent registry.
#[derive(Debug, Clone)]
pub(crate) struct InternalWorkerSessionSnapshot {
pub entries: Vec<LogEntry>,
pub status: WorkerStatus,
pub error: Option<String>,
pub in_flight: InFlightSnapshot,
pub internal_workers: Vec<protocol::InternalWorkerSnapshot>,
}
#[derive(Clone)]
pub(crate) struct InternalWorkerSessionHandle {
command_tx: tokio::sync::mpsc::Sender<InternalWorkerSessionCommand>,
@@ -256,6 +288,12 @@ pub(crate) struct InternalWorkerSessionHandle {
session_id: SessionId,
segment_id: SegmentId,
state_changed: Arc<tokio::sync::Notify>,
in_flight: InFlightEvents,
event_tx: broadcast::Sender<Event>,
visibility: InternalWorkerVisibility,
last_error: Arc<Mutex<Option<String>>>,
child_registry: Option<Arc<SpawnedWorkerRegistry>>,
sink: SegmentLogSink,
}
impl InternalWorkerSessionHandle {
@@ -267,6 +305,54 @@ impl InternalWorkerSessionHandle {
InternalWorkerSessionStatus::decode(self.status.load(std::sync::atomic::Ordering::Acquire))
}
pub(crate) fn visibility(&self) -> InternalWorkerVisibility {
self.visibility
}
pub(crate) fn subscribe_events(&self) -> broadcast::Receiver<Event> {
self.event_tx.subscribe()
}
pub(crate) fn protocol_sender(&self) -> broadcast::Sender<Event> {
self.event_tx.clone()
}
#[cfg(test)]
pub(crate) fn publish_test_entry(&self, entry: LogEntry) {
self.sink.publish(entry);
}
#[cfg(test)]
pub(crate) fn emit_test_text_delta(&self, text: &str) {
let block_id = self.in_flight.start_text_block();
self.in_flight.text_delta(block_id, text.to_owned());
}
pub(crate) fn protocol_snapshot(&self) -> InternalWorkerSessionSnapshot {
let (entries, in_flight) = {
let guard = self.in_flight.snapshot_guard();
let (entries, _) = self.sink.subscribe_with_snapshot();
(entries, snapshot_from_guard(&guard))
};
InternalWorkerSessionSnapshot {
entries,
status: match self.status() {
InternalWorkerSessionStatus::Running => WorkerStatus::Running,
InternalWorkerSessionStatus::Idle => WorkerStatus::Idle,
InternalWorkerSessionStatus::Stopping
| InternalWorkerSessionStatus::Stopped
| InternalWorkerSessionStatus::Failed => WorkerStatus::Paused,
},
error: self.last_error.lock().unwrap().clone(),
in_flight,
internal_workers: self
.child_registry
.as_ref()
.map(|registry| registry.internal_worker_snapshots())
.unwrap_or_default(),
}
}
pub(crate) fn entries(&self) -> Vec<LogEntry> {
self.store
.read_all(self.session_id, self.segment_id)
@@ -305,8 +391,17 @@ impl InternalWorkerSessionHandle {
std::sync::atomic::Ordering::Release,
);
self.state_changed.notify_waiters();
let message = "internal Worker session actor is unavailable".to_owned();
*self.last_error.lock().unwrap() = Some(message.clone());
let _ = self.event_tx.send(Event::Error {
code: protocol::ErrorCode::Internal,
message,
});
return Err(InternalWorkerSessionError::Unavailable);
}
let _ = self.event_tx.send(Event::Status {
status: WorkerStatus::Running,
});
Ok(())
}
@@ -423,11 +518,48 @@ pub(crate) async fn spawn_internal_worker_session(
spawn_prepared_internal_worker_session(worker, store, input, None).await
}
fn spawn_internal_log_event_bridge(sink: SegmentLogSink, event_tx: broadcast::Sender<Event>) {
let (_, mut log_rx) = sink.subscribe_with_snapshot();
tokio::spawn(async move {
loop {
match log_rx.recv().await {
Ok(entry) => {
if let Some(event) = live_log_entry_event(entry) {
let _ = event_tx.send(event);
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
let _ = event_tx.send(Event::Error {
code: protocol::ErrorCode::Internal,
message: format!(
"internal Worker session-log output lagged by {skipped} entries; reconnect to resynchronize"
),
});
break;
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
});
}
pub(crate) async fn prepare_internal_worker_session(
mut worker: Worker<Box<dyn LlmClient>, EphemeralSessionStore>,
store: EphemeralSessionStore,
visibility: InternalWorkerVisibility,
child_registry: Option<Arc<SpawnedWorkerRegistry>>,
on_turn_end: Option<Arc<dyn Fn(InternalWorkerSessionStatus) + Send + Sync>>,
) -> Result<InternalWorkerSessionHandle, InternalWorkerSessionError> {
let (event_tx, _event_rx) = broadcast::channel(256);
let sink = worker.sink();
spawn_internal_log_event_bridge(sink.clone(), event_tx.clone());
let alerter = Alerter::new(event_tx.clone());
let in_flight = InFlightEvents::new(event_tx.clone());
worker.attach_alerter(alerter.clone());
worker.attach_event_tx(event_tx.clone());
worker.attach_in_flight_events(in_flight.clone());
wire_event_bridges_on_engine(&mut worker, &event_tx, &alerter, &in_flight);
let session_id = worker.session_id();
let segment_id = worker.segment_id();
let (command_tx, mut command_rx) = tokio::sync::mpsc::channel(8);
@@ -435,6 +567,7 @@ pub(crate) async fn prepare_internal_worker_session(
InternalWorkerSessionStatus::Idle.encode(),
));
let state_changed = Arc::new(tokio::sync::Notify::new());
let last_error = Arc::new(Mutex::new(None));
let handle = InternalWorkerSessionHandle {
command_tx,
status: status.clone(),
@@ -442,6 +575,12 @@ pub(crate) async fn prepare_internal_worker_session(
session_id,
segment_id,
state_changed: state_changed.clone(),
in_flight,
event_tx: event_tx.clone(),
visibility,
last_error: last_error.clone(),
child_registry,
sink,
};
tokio::spawn(async move {
@@ -453,11 +592,25 @@ pub(crate) async fn prepare_internal_worker_session(
loop {
tokio::select! {
result = &mut run => {
let turn_status = match result {
Ok(_) => InternalWorkerSessionStatus::Idle,
Err(_) => InternalWorkerSessionStatus::Failed,
let (turn_status, error) = match result {
Ok(_) => (InternalWorkerSessionStatus::Idle, None),
Err(error) => (
InternalWorkerSessionStatus::Failed,
Some(error.to_string()),
),
};
status.store(turn_status.encode(), std::sync::atomic::Ordering::Release);
if let Some(message) = error {
*last_error.lock().unwrap() = Some(message.clone());
let _ = event_tx.send(Event::Error {
code: protocol::ErrorCode::Internal,
message,
});
} else {
let _ = event_tx.send(Event::Status {
status: WorkerStatus::Idle,
});
}
if let Some(callback) = &on_turn_end {
callback(turn_status);
}
@@ -470,6 +623,8 @@ pub(crate) async fn prepare_internal_worker_session(
let _ = cancel_sender.send(()).await;
let _ = (&mut run).await;
status.store(InternalWorkerSessionStatus::Stopped.encode(), std::sync::atomic::Ordering::Release);
let _ = event_tx.send(Event::Status { status: WorkerStatus::Paused });
let _ = event_tx.send(Event::Shutdown);
state_changed.notify_waiters();
let _ = done.send(());
return;
@@ -491,6 +646,10 @@ pub(crate) async fn prepare_internal_worker_session(
InternalWorkerSessionStatus::Stopped.encode(),
std::sync::atomic::Ordering::Release,
);
let _ = event_tx.send(Event::Status {
status: WorkerStatus::Paused,
});
let _ = event_tx.send(Event::Shutdown);
state_changed.notify_waiters();
let _ = done.send(());
return;
@@ -509,7 +668,14 @@ pub(crate) async fn spawn_prepared_internal_worker_session(
input: String,
on_turn_end: Option<Arc<dyn Fn(InternalWorkerSessionStatus) + Send + Sync>>,
) -> Result<InternalWorkerSessionHandle, InternalWorkerSessionError> {
let handle = prepare_internal_worker_session(worker, store, on_turn_end).await?;
let handle = prepare_internal_worker_session(
worker,
store,
InternalWorkerVisibility::ServicePrivate,
None,
on_turn_end,
)
.await?;
handle.send(input).await?;
Ok(handle)
}
@@ -709,6 +875,37 @@ impl session_store::WorkerMetadataStore for EphemeralSessionStore {
}
}
#[cfg(test)]
pub(crate) fn test_internal_worker_session(
visibility: InternalWorkerVisibility,
) -> (InternalWorkerSessionHandle, broadcast::Sender<Event>) {
let store = EphemeralSessionStore::default();
let session_id = session_store::new_session_id();
let segment_id = session_store::new_segment_id();
let (command_tx, mut command_rx) = tokio::sync::mpsc::channel(1);
tokio::spawn(async move { while command_rx.recv().await.is_some() {} });
let (event_tx, _) = broadcast::channel(256);
let sink = SegmentLogSink::new();
spawn_internal_log_event_bridge(sink.clone(), event_tx.clone());
let handle = InternalWorkerSessionHandle {
command_tx,
status: Arc::new(std::sync::atomic::AtomicU8::new(
InternalWorkerSessionStatus::Idle.encode(),
)),
store,
session_id,
segment_id,
state_changed: Arc::new(tokio::sync::Notify::new()),
in_flight: InFlightEvents::new(event_tx.clone()),
event_tx: event_tx.clone(),
visibility,
last_error: Arc::new(Mutex::new(None)),
child_registry: None,
sink,
};
(handle, event_tx)
}
#[cfg(test)]
mod tests {
use std::pin::Pin;
+14
View File
@@ -595,6 +595,20 @@ mod tests {
assert!(!catalog.projection.templates.is_empty());
}
#[test]
fn orchestrator_prompt_uses_normal_branch_integration_before_escalation() {
let catalog = PromptCatalog::builtins_only().unwrap();
let prompt = &catalog.projection.templates["role.orchestrator"];
assert!(prompt.contains("`merge_from`"));
assert!(prompt.contains("`merge_to`"));
assert!(prompt.contains("normal source-control operations"));
assert!(prompt.contains("switch to `merge_to`"));
assert!(prompt.contains("concrete source-control or provider failure"));
assert!(prompt.contains("does not update the branch itself"));
assert!(!prompt.contains("use the Ticket repository `origin` transport"));
}
#[test]
fn graph_rejects_dynamic_legacy_missing_and_cycles() {
let invalid = BTreeMap::from([
+1
View File
@@ -284,6 +284,7 @@ mod tests {
},
status: WorkerStatus::Idle,
in_flight: Default::default(),
internal_workers: Vec::new(),
}
}
+287 -4
View File
@@ -10,17 +10,20 @@
use std::collections::HashSet;
use std::io;
use std::sync::{
Arc,
atomic::{AtomicBool, Ordering},
Arc, Mutex,
atomic::{AtomicBool, AtomicU64, Ordering},
};
use manifest::{Permission, ScopeRule, SharedScope};
use protocol::{Event, InternalWorkerKind, InternalWorkerRef, InternalWorkerSnapshot};
use session_store::{
WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerStoreError,
};
use tokio::sync::broadcast;
use tracing::warn;
use workdir::WorkdirDelegation;
use crate::internal_worker::InternalWorkerSessionHandle;
use crate::internal_worker::{InternalWorkerSessionHandle, InternalWorkerVisibility};
use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord};
use crate::runtime::worker_allocation;
@@ -28,21 +31,33 @@ use crate::runtime::worker_allocation;
pub(crate) struct InternalSpawnedWorkerRecord {
pub worker_name: String,
pub scope_delegated: Vec<ScopeRule>,
pub workdir_delegation: Arc<WorkdirDelegation>,
#[cfg(test)]
pub installed_tools: Arc<[String]>,
pub session: InternalWorkerSessionHandle,
scope_reclaimed: Arc<AtomicBool>,
protocol_revision: Arc<AtomicU64>,
forwarding_started: Arc<AtomicBool>,
}
impl InternalSpawnedWorkerRecord {
pub(crate) fn new(
worker_name: String,
scope_delegated: Vec<ScopeRule>,
workdir_delegation: WorkdirDelegation,
#[cfg(test)] installed_tools: Vec<String>,
session: InternalWorkerSessionHandle,
) -> Self {
Self {
worker_name,
scope_delegated,
workdir_delegation: Arc::new(workdir_delegation),
#[cfg(test)]
installed_tools: installed_tools.into(),
session,
scope_reclaimed: Arc::new(AtomicBool::new(false)),
protocol_revision: Arc::new(AtomicU64::new(0)),
forwarding_started: Arc::new(AtomicBool::new(false)),
}
}
@@ -53,6 +68,19 @@ impl InternalSpawnedWorkerRecord {
fn restore_scope_reclaim(&self) {
self.scope_reclaimed.store(false, Ordering::Release);
}
fn protocol_ref(&self, parent_session_id: Option<String>) -> InternalWorkerRef {
InternalWorkerRef {
session_id: self.session.session_id_string(),
name: self.worker_name.clone(),
parent_session_id,
kind: InternalWorkerKind::SubWorker,
}
}
fn protocol_revision(&self) -> u64 {
self.protocol_revision.load(Ordering::Acquire)
}
}
pub(crate) struct InternalSpawnReservation {
@@ -73,7 +101,8 @@ impl InternalSpawnReservation {
.internal_records
.lock()
.map_err(|_| io::Error::other("internal spawned-worker registry lock poisoned"))?
.push(record);
.push(record.clone());
self.registry.start_protocol_forwarding(record);
self.committed = true;
Ok(())
}
@@ -93,6 +122,7 @@ pub struct SpawnedWorkerRegistry {
internal_records: std::sync::Mutex<Vec<InternalSpawnedWorkerRecord>>,
internal_names: std::sync::Mutex<HashSet<String>>,
parent_scope: Option<SharedScope>,
parent_protocol: Mutex<Option<(broadcast::Sender<Event>, String)>>,
}
pub struct SpawnedWorkerRegistryLoad {
@@ -108,6 +138,7 @@ impl SpawnedWorkerRegistry {
internal_records: std::sync::Mutex::new(Vec::new()),
internal_names: std::sync::Mutex::new(HashSet::new()),
parent_scope: None,
parent_protocol: Mutex::new(None),
})
}
@@ -116,6 +147,7 @@ impl SpawnedWorkerRegistry {
internal_records: std::sync::Mutex::new(Vec::new()),
internal_names: std::sync::Mutex::new(HashSet::new()),
parent_scope: Some(parent_scope),
parent_protocol: Mutex::new(None),
})
}
@@ -193,6 +225,7 @@ impl SpawnedWorkerRegistry {
internal_records: std::sync::Mutex::new(Vec::new()),
internal_names: std::sync::Mutex::new(HashSet::new()),
parent_scope,
parent_protocol: Mutex::new(None),
}),
reclaimed_unreachable: !persisted_children.is_empty(),
})
@@ -220,6 +253,96 @@ impl SpawnedWorkerRegistry {
})
}
pub(crate) fn attach_parent_protocol(
&self,
event_tx: broadcast::Sender<Event>,
parent_session_id: String,
) {
*self.parent_protocol.lock().unwrap() = Some((event_tx, parent_session_id));
for record in self.internal_records.lock().unwrap().clone() {
self.start_protocol_forwarding(record);
}
}
fn start_protocol_forwarding(&self, record: InternalSpawnedWorkerRecord) {
if record.session.visibility() != InternalWorkerVisibility::ParentClient
|| record.forwarding_started.swap(true, Ordering::AcqRel)
{
return;
}
let Some((parent_tx, parent_session_id)) = self.parent_protocol.lock().unwrap().clone()
else {
record.forwarding_started.store(false, Ordering::Release);
return;
};
let worker = record.protocol_ref(Some(parent_session_id));
let protocol_revision = record.protocol_revision.clone();
let mut child_rx = record.session.subscribe_events();
tokio::spawn(async move {
loop {
match child_rx.recv().await {
Ok(event) => {
let shutdown = matches!(event, Event::Shutdown);
let revision = protocol_revision.fetch_add(1, Ordering::AcqRel) + 1;
let _ = parent_tx.send(Event::InternalWorker {
worker: worker.clone(),
revision,
event: Box::new(event),
});
if shutdown {
break;
}
}
Err(broadcast::error::RecvError::Lagged(skipped)) => {
let revision = protocol_revision.fetch_add(1, Ordering::AcqRel) + 1;
let _ = parent_tx.send(Event::InternalWorker {
worker: worker.clone(),
revision,
event: Box::new(Event::Error {
code: protocol::ErrorCode::Internal,
message: format!(
"internal Worker output lagged by {skipped} events; reconnect to resynchronize"
),
}),
});
}
Err(broadcast::error::RecvError::Closed) => break,
}
}
});
}
pub(crate) fn internal_worker_snapshots(&self) -> Vec<InternalWorkerSnapshot> {
let parent_session_id = self
.parent_protocol
.lock()
.unwrap()
.as_ref()
.map(|(_, id)| id.clone());
self.internal_records
.lock()
.unwrap()
.iter()
.filter(|record| record.session.visibility() == InternalWorkerVisibility::ParentClient)
.map(|record| {
let snapshot = record.session.protocol_snapshot();
InternalWorkerSnapshot {
worker: record.protocol_ref(parent_session_id.clone()),
revision: record.protocol_revision(),
entries: snapshot
.entries
.into_iter()
.filter_map(|entry| serde_json::to_value(entry).ok())
.collect(),
status: snapshot.status,
error: snapshot.error,
in_flight: snapshot.in_flight,
internal_workers: snapshot.internal_workers,
}
})
.collect()
}
pub(crate) fn get_internal(&self, worker_name: &str) -> Option<InternalSpawnedWorkerRecord> {
self.internal_records
.lock()
@@ -247,6 +370,7 @@ impl SpawnedWorkerRegistry {
if !record.claim_scope_reclaim() {
return Ok(false);
}
record.workdir_delegation.release();
let result = if let Some(parent_scope) = &self.parent_scope {
parent_scope
.update(|current| current.with_removed_deny_rules(delegated_write_rules(record)))
@@ -387,3 +511,162 @@ fn record_from_worker_state(child: &WorkerSpawnedChild) -> io::Result<SpawnedWor
fn store_error_to_io(error: WorkerStoreError) -> io::Error {
io::Error::other(error)
}
#[cfg(test)]
mod tests {
use std::time::Duration;
use manifest::{Scope, ScopeConfig};
use session_store::LogEntry;
use super::*;
use crate::internal_worker::test_internal_worker_session;
fn registry() -> Arc<SpawnedWorkerRegistry> {
let scope = Scope::from_config(&ScopeConfig {
allow: vec![ScopeRule {
target: std::path::PathBuf::from("/tmp"),
permission: Permission::Read,
recursive: true,
}],
deny: Vec::new(),
})
.unwrap();
SpawnedWorkerRegistry::new_internal("parent".into(), SharedScope::new(scope))
}
async fn record(
name: &str,
visibility: InternalWorkerVisibility,
) -> (InternalSpawnedWorkerRecord, broadcast::Sender<Event>) {
let (session, sender) = test_internal_worker_session(visibility);
let root = std::path::PathBuf::from("/tmp");
let scope = Scope::from_config(&ScopeConfig {
allow: vec![ScopeRule {
target: root.clone(),
permission: Permission::Read,
recursive: true,
}],
deny: Vec::new(),
})
.unwrap();
let source = workdir::delegation_capable_session(Arc::new(
workdir::LocalWorkdirSession::materialized_bound(
workdir::Workdir::new("registry-test"),
root.clone(),
root,
SharedScope::new(scope),
workdir::WorkdirSessionCapabilities::ALL,
),
));
let delegation = source
.delegate(workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: workdir::WorkdirPath::new("").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: true,
}],
cwd: workdir::WorkdirPath::new("").unwrap(),
})
.await
.unwrap();
(
InternalSpawnedWorkerRecord::new(
name.into(),
Vec::new(),
delegation,
Vec::new(),
session,
),
sender,
)
}
#[tokio::test]
async fn visible_internal_output_is_wrapped_after_registry_insertion() {
let registry = registry();
let (parent_tx, mut parent_rx) = broadcast::channel(16);
registry.attach_parent_protocol(parent_tx, "parent-session".into());
let (record, child_tx) = record("research", InternalWorkerVisibility::ParentClient).await;
registry
.internal_records
.lock()
.unwrap()
.push(record.clone());
registry.start_protocol_forwarding(record.clone());
child_tx
.send(Event::TextDone {
text: "answer".into(),
})
.unwrap();
let event = tokio::time::timeout(Duration::from_secs(1), parent_rx.recv())
.await
.unwrap()
.unwrap();
assert!(matches!(
event,
Event::InternalWorker { worker, revision: 1, event }
if worker.name == "research"
&& worker.parent_session_id.as_deref() == Some("parent-session")
&& matches!(*event, Event::TextDone { ref text } if text == "answer")
));
record.session.publish_test_entry(LogEntry::UserInput {
ts: 1,
segments: vec![protocol::Segment::text("question")],
extensions: Vec::new(),
});
let committed = tokio::time::timeout(Duration::from_secs(1), parent_rx.recv())
.await
.unwrap()
.unwrap();
assert!(matches!(
committed,
Event::InternalWorker { revision: 2, event, .. }
if matches!(*event, Event::UserMessage { .. })
));
let snapshots = registry.internal_worker_snapshots();
assert_eq!(snapshots.len(), 1);
assert_eq!(snapshots[0].revision, 2);
assert_eq!(snapshots[0].entries.len(), 1);
record.session.emit_test_text_delta("partial");
let streamed = tokio::time::timeout(Duration::from_secs(1), parent_rx.recv())
.await
.unwrap()
.unwrap();
assert!(matches!(
streamed,
Event::InternalWorker { revision: 3, event, .. }
if matches!(*event, Event::TextDelta { ref text } if text == "partial")
));
let snapshots = registry.internal_worker_snapshots();
assert_eq!(snapshots[0].revision, 3);
assert_eq!(snapshots[0].in_flight.blocks.len(), 1);
}
#[tokio::test]
async fn service_private_internal_output_is_never_disclosed() {
let registry = registry();
let (parent_tx, mut parent_rx) = broadcast::channel(16);
registry.attach_parent_protocol(parent_tx, "parent-session".into());
let (record, child_tx) =
record("memory-helper", InternalWorkerVisibility::ServicePrivate).await;
registry
.internal_records
.lock()
.unwrap()
.push(record.clone());
registry.start_protocol_forwarding(record);
let _ = child_tx.send(Event::TextDone {
text: "private".into(),
});
assert!(
tokio::time::timeout(Duration::from_millis(50), parent_rx.recv())
.await
.is_err()
);
assert!(registry.internal_worker_snapshots().is_empty());
}
}
+182 -226
View File
@@ -10,21 +10,27 @@ use std::sync::Arc;
use arc_swap::ArcSwap;
use async_trait::async_trait;
use fs_operation::FsPath;
use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
use manifest::{
CompactionConfigPartial, DelegationScope, EngineManifestConfig, FileUploadLimitsPartial,
Permission, PermissionConfigPartial, ProfileDiscovery, ProfileError, ProfileRegistry,
ProfileRegistrySource, ProfileResolveOptions, ProfileResolver, ProfileSelector, Scope,
ScopeConfig, ScopeRule, SessionConfigPartial, SharedScope, ToolOutputLimitsPartial,
WorkerManifest, WorkerManifestConfig, WorkerMetaConfig,
CompactionConfigPartial, EngineManifestConfig, FileUploadLimitsPartial,
PermissionConfigPartial, ProfileDiscovery, ProfileError, ProfileRegistry,
ProfileRegistrySource, ProfileResolveOptions, ProfileResolver, ProfileSelector, ScopeConfig,
ScopeRule, SessionConfigPartial, ToolOutputLimitsPartial, WorkerManifest, WorkerManifestConfig,
WorkerMetaConfig,
};
use serde::Deserialize;
use tokio::sync::mpsc;
use workdir::{
WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule,
WorkdirSessionHandle,
};
use crate::PromptCatalogSource;
use crate::controller::register_worker_tools;
use crate::internal_worker::{
EphemeralSessionStore, InternalWorkerSessionStatus, prepare_internal_worker_session,
EphemeralSessionStore, InternalWorkerSessionStatus, InternalWorkerVisibility,
prepare_internal_worker_session,
};
use crate::prompt::catalog::PromptCatalog;
use crate::spawn::registry::SpawnedWorkerRegistry;
@@ -48,11 +54,10 @@ struct SubWorkerSpawnInput {
/// Exact catalog-root dotted Prompt name (for example `default` or `role.coder`).
#[serde(default)]
instruction: Option<String>,
/// Child process/tool working directory. This is not the runtime workspace
/// root and grants no filesystem authority. When omitted, the spawned SubWorker
/// starts in the spawner's current working directory.
/// Logical Workdir-relative child tool working directory. This path is not
/// a host path and grants no authority. When omitted, the Workdir root is used.
#[serde(default)]
cwd: Option<PathBuf>,
cwd: Option<String>,
/// First message sent to the spawned SubWorker via `Method::Run`.
task: String,
/// Allow rules delegated to the spawned SubWorker. Must be a subset of the
@@ -72,8 +77,9 @@ struct ReviewerHandoffInput {
#[derive(Debug, Deserialize, schemars::JsonSchema)]
struct ScopeRuleInput {
/// Absolute target path. Relative paths are rejected.
target: PathBuf,
/// Logical Workdir-relative target such as `.` or `src`. Absolute host
/// paths and parent traversal are rejected.
target: String,
/// `"read"` or `"write"`.
permission: PermissionInput,
/// When `false`, the rule matches the target itself and its direct
@@ -93,15 +99,6 @@ fn default_true() -> bool {
true
}
impl From<PermissionInput> for Permission {
fn from(p: PermissionInput) -> Self {
match p {
PermissionInput::Read => Permission::Read,
PermissionInput::Write => Permission::Write,
}
}
}
#[derive(Debug, Clone)]
struct AvailableProfiles {
registry: Option<ProfileRegistry>,
@@ -269,7 +266,8 @@ pub struct SubWorkerSpawnTool {
workspace_root: PathBuf,
/// Directory the spawned SubWorker's tools should use when the LLM did not
/// override it. Defaults to the spawner's cwd.
spawner_cwd: PathBuf,
/// Active provider-backed Workdir session from which child leases are captured.
source_workdir_session: Option<WorkdirSessionHandle>,
/// Parent-owned in-memory registry shared by the five SubWorker tools.
registry: Arc<SpawnedWorkerRegistry>,
/// Spawner's resolved Manifest. `profile = "inherit"` derives the
@@ -279,18 +277,6 @@ pub struct SubWorkerSpawnTool {
prompt_loader: PromptCatalogSource,
/// Compact selector list shared by tool description and diagnostics.
available_profiles: AvailableProfiles,
/// Spawner's runtime scope. After a successful spawn, the
/// `Permission::Write` rules in the delegated scope are revoked
/// from the spawner's in-memory view (a `deny(Write, target)` is
/// pushed on top, downgrading the spawner's effective access on
/// those paths to `Read`). Mirrors the worker-allocation's
/// `effective_write` semantics: Write is the only permission
/// tracked across Workers, so revocation only touches Write.
spawner_scope: SharedScope,
/// Filesystem scope this Worker is allowed to subdelegate to children.
/// This is intentionally separate from `spawner_scope`, which authorizes
/// the current Worker's own direct tools.
delegation_scope: DelegationScope,
internal_client_override: Option<Box<dyn llm_engine::llm_client::LlmClient>>,
}
@@ -307,13 +293,11 @@ impl SubWorkerSpawnTool {
parent_notifications: ParentNotificationTarget,
runtime_base: PathBuf,
workspace_root: PathBuf,
spawner_cwd: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>,
registry: Arc<SpawnedWorkerRegistry>,
spawner_manifest: WorkerManifest,
prompt_loader: PromptCatalogSource,
available_profiles: AvailableProfiles,
spawner_scope: SharedScope,
delegation_scope: DelegationScope,
) -> Self {
Self {
spawner_name,
@@ -321,13 +305,11 @@ impl SubWorkerSpawnTool {
parent_notifications,
runtime_base,
workspace_root,
spawner_cwd,
source_workdir_session,
registry,
spawner_manifest,
prompt_loader,
available_profiles,
spawner_scope,
delegation_scope,
internal_client_override: None,
}
}
@@ -385,9 +367,16 @@ impl Tool for SubWorkerSpawnTool {
.reserve_internal_name(input.name.clone())
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
let scope_allow = parse_scope(&input.scope)?;
self.validate_delegation_scope(&scope_allow)?;
let child_cwd = validate_spawn_cwd(input.cwd.as_deref(), &scope_allow, &self.spawner_cwd)?;
let workdir_rules = parse_workdir_scope(&input.scope)?;
let source_workdir_session =
require_active_workdir_session(self.source_workdir_session.as_ref())?;
let delegation_request = workdir_delegation_request(input.cwd.as_deref(), workdir_rules)?;
let workdir_delegation = source_workdir_session
.delegate(delegation_request)
.await
.map_err(|error| {
ToolError::InvalidArgument(format!("delegate Workdir session: {error}"))
})?;
let spawn_selector =
parse_spawn_profile_selector(input.profile.as_deref()).map_err(|msg| {
@@ -396,6 +385,7 @@ impl Tool for SubWorkerSpawnTool {
self.available_profiles.error_suffix()
))
})?;
let scope_allow = Vec::new();
let spawn_config_json = self
.build_spawn_config_json(
&input.name,
@@ -413,11 +403,14 @@ impl Tool for SubWorkerSpawnTool {
allow: scope_allow.clone(),
deny: Vec::new(),
};
let child_manifest =
let mut child_manifest =
WorkerManifest::try_from(WorkerManifestConfig::builtin_defaults().merge(child_config))
.map_err(|error| {
ToolError::ExecutionFailed(format!("resolve child manifest: {error}"))
})?;
// Delegated children stay bound to their scoped session and cannot use
// Workspace attachment tools to replace it with parent-level authority.
child_manifest.feature.manage_workdir.enabled = false;
let reviewer_capability = input.review.as_ref().map(|review| {
(
review.ticket_id.clone(),
@@ -458,8 +451,7 @@ impl Tool for SubWorkerSpawnTool {
self.workspace_context.clone()
};
let store = EphemeralSessionStore::default();
let filesystem_authority =
WorkerFilesystemAuthority::local(self.workspace_root.clone(), child_cwd.clone());
let filesystem_authority = WorkerFilesystemAuthority::None;
let mut child = Worker::<Box<dyn llm_engine::llm_client::LlmClient>, EphemeralSessionStore>::from_internal_manifest_with_context(
child_manifest,
store.clone(),
@@ -472,6 +464,7 @@ impl Tool for SubWorkerSpawnTool {
)
.await
.map_err(|error| ToolError::ExecutionFailed(format!("build Internal Worker: {error}")))?;
child.bind_workdir_session(Some(workdir_delegation.scoped_session.clone()));
let child_scope = child.scope().clone();
let child_registry = SpawnedWorkerRegistry::new_internal(input.name.clone(), child_scope);
register_worker_tools(
@@ -481,36 +474,29 @@ impl Tool for SubWorkerSpawnTool {
.join(&input.name)
.join("bash-output"),
self.runtime_base.clone(),
child_registry,
child_registry.clone(),
None,
)
.await
.map_err(|error| {
ToolError::ExecutionFailed(format!("install Internal Worker features: {error}"))
})?;
// Transfer delegated Write authority before the child accepts its first turn. This closes
// the parallel-tool window where parent and child could otherwise both write the same path.
// The machine-wide allocation remains owned by the parent Worker; no fake child PID/socket
// identity is introduced.
let revoke_write: Vec<ScopeRule> = scope_allow
.iter()
.filter(|rule| rule.permission == Permission::Write)
.cloned()
#[cfg(test)]
let installed_tools = child
.engine()
.tool_server_handle()
.tool_definitions_sorted()
.into_iter()
.map(|definition| definition.name)
.collect();
if !revoke_write.is_empty() {
self.spawner_scope
.update(|current| current.with_added_deny_rules(revoke_write.clone()))
.map_err(|error| {
ToolError::ExecutionFailed(format!("revoke spawner scope: {error}"))
})?;
}
let child_name = input.name.clone();
let registry = Arc::downgrade(&self.registry);
let parent_notifications = self.parent_notifications.clone();
let session_result = prepare_internal_worker_session(
child,
store,
InternalWorkerVisibility::ParentClient,
Some(child_registry.clone()),
Some(Arc::new(move |status| {
if status == InternalWorkerSessionStatus::Failed {
if let Some(registry) = registry.upgrade() {
@@ -530,19 +516,11 @@ impl Tool for SubWorkerSpawnTool {
})),
)
.await;
let session = match session_result {
Ok(session) => session,
Err(error) => {
if !revoke_write.is_empty() {
let _ = self
.spawner_scope
.update(|current| current.with_removed_deny_rules(revoke_write.clone()));
}
return Err(ToolError::ExecutionFailed(format!(
"prepare Internal Worker session: {error}"
)));
}
};
let session = session_result.map_err(|error| {
ToolError::ExecutionFailed(format!("prepare Internal Worker session: {error}"))
})?;
child_registry
.attach_parent_protocol(session.protocol_sender(), session.session_id_string());
if let Some((ticket_id, capability_token)) = &reviewer_capability {
let workspace_id = self.workspace_context.workspace_id().ok_or_else(|| {
@@ -605,15 +583,13 @@ impl Tool for SubWorkerSpawnTool {
let record = crate::spawn::registry::InternalSpawnedWorkerRecord::new(
input.name.clone(),
scope_allow,
workdir_delegation,
#[cfg(test)]
installed_tools,
session.clone(),
);
if let Err(error) = name_reservation.commit(record) {
let _ = session.stop().await;
if !revoke_write.is_empty() {
let _ = self
.spawner_scope
.update(|current| current.with_removed_deny_rules(revoke_write));
}
return Err(ToolError::ExecutionFailed(format!(
"register Internal Worker session: {error}"
)));
@@ -635,107 +611,68 @@ impl Tool for SubWorkerSpawnTool {
}
}
impl SubWorkerSpawnTool {
fn validate_delegation_scope(&self, scope_allow: &[ScopeRule]) -> Result<(), ToolError> {
if self.delegation_scope.is_empty() && !scope_allow.is_empty() {
return Err(ToolError::InvalidArgument(
"SubWorkerSpawn requires delegation authority, but this Worker has no delegation scope grant; direct filesystem scope only authorizes this Worker's own tools".into(),
));
}
for rule in scope_allow {
let allowed = self
.delegation_scope
.allows_rule(rule)
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
if !allowed {
fn logical_workdir_path(value: &str, field: &str) -> Result<FsPath, ToolError> {
let path = Path::new(value);
if path.is_absolute() {
return Err(ToolError::InvalidArgument(format!(
"requested child scope {} {:?} is outside this Worker's delegation scope grant",
rule.target.display(),
rule.permission
"{field} must be Workdir-relative, got `{value}`"
)));
}
}
Ok(())
}
let normalized = path
.components()
.filter_map(|component| match component {
std::path::Component::CurDir => None,
other => Some(other.as_os_str()),
})
.collect::<PathBuf>();
let normalized = normalized.to_str().ok_or_else(|| {
ToolError::InvalidArgument(format!("{field} `{value}` is not valid UTF-8"))
})?;
FsPath::new(normalized).map_err(|error| {
ToolError::InvalidArgument(format!(
"{field} `{value}` is not a valid logical Workdir path: {error}"
))
})
}
fn parse_scope(rules: &[ScopeRuleInput]) -> Result<Vec<ScopeRule>, ToolError> {
fn parse_workdir_scope(rules: &[ScopeRuleInput]) -> Result<Vec<WorkdirDelegationRule>, ToolError> {
if rules.is_empty() {
return Err(ToolError::InvalidArgument("scope must not be empty".into()));
}
rules
.iter()
.map(|r| {
if !r.target.is_absolute() {
return Err(ToolError::InvalidArgument(format!(
"scope.target must be absolute: {}",
r.target.display()
)));
}
Ok(ScopeRule {
target: r.target.clone(),
permission: r.permission.into(),
recursive: r.recursive,
.map(|rule| {
Ok(WorkdirDelegationRule {
target: logical_workdir_path(&rule.target, "scope.target")?,
permission: match rule.permission {
PermissionInput::Read => WorkdirDelegationPermission::Read,
PermissionInput::Write => WorkdirDelegationPermission::Write,
},
recursive: rule.recursive,
})
})
.collect()
}
fn validate_spawn_cwd(
cwd: Option<&Path>,
scope_allow: &[ScopeRule],
default_cwd: &Path,
) -> Result<PathBuf, ToolError> {
let Some(cwd) = cwd else {
return Ok(default_cwd.to_path_buf());
};
if !cwd.is_absolute() {
return Err(ToolError::InvalidArgument(format!(
"SubWorkerSpawn.cwd must be absolute: {}",
cwd.display()
)));
}
let metadata = std::fs::metadata(cwd).map_err(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
ToolError::InvalidArgument(format!(
"SubWorkerSpawn.cwd does not exist: {}",
cwd.display()
))
} else {
ToolError::InvalidArgument(format!(
"SubWorkerSpawn.cwd is not usable: {}: {e}",
cwd.display()
))
}
})?;
if !metadata.is_dir() {
return Err(ToolError::InvalidArgument(format!(
"SubWorkerSpawn.cwd must be a directory: {}",
cwd.display()
)));
}
let canonical = std::fs::canonicalize(cwd).map_err(|e| {
ToolError::InvalidArgument(format!(
"SubWorkerSpawn.cwd is not usable: {}: {e}",
cwd.display()
))
})?;
let child_scope = Scope::from_config(&ScopeConfig {
allow: scope_allow.to_vec(),
deny: Vec::new(),
fn workdir_delegation_request(
cwd: Option<&str>,
rules: Vec<WorkdirDelegationRule>,
) -> Result<WorkdirDelegationRequest, ToolError> {
Ok(WorkdirDelegationRequest {
rules,
cwd: logical_workdir_path(cwd.unwrap_or("."), "cwd")?,
})
.map_err(|e| {
ToolError::InvalidArgument(format!(
"requested child scope cannot validate SubWorkerSpawn.cwd: {e}"
))
})?;
if !child_scope.is_readable(&canonical) {
return Err(ToolError::InvalidArgument(format!(
"SubWorkerSpawn.cwd {} is outside the child's delegated readable scope; cwd grants no authority, so add an explicit read or write scope rule covering it",
cwd.display()
)));
}
Ok(canonical)
fn require_active_workdir_session(
session: Option<&WorkdirSessionHandle>,
) -> Result<&WorkdirSessionHandle, ToolError> {
session.ok_or_else(|| {
ToolError::InvalidArgument(
"SubWorkerSpawn requires an active Workdir session; attach a Workdir before delegating filesystem access"
.to_string(),
)
})
}
/// Serialise the internal manifest config that gets handed to the child
@@ -943,10 +880,9 @@ pub(crate) fn sub_worker_spawn_tool(
parent_notifications: ParentNotificationTarget,
runtime_base: PathBuf,
workspace_root: PathBuf,
spawner_cwd: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>,
registry: Arc<SpawnedWorkerRegistry>,
spawner_manifest: WorkerManifest,
spawner_scope: SharedScope,
prompts: Arc<ArcSwap<PromptCatalog>>,
) -> ToolDefinition {
sub_worker_spawn_tool_impl(
@@ -955,10 +891,9 @@ pub(crate) fn sub_worker_spawn_tool(
parent_notifications,
runtime_base,
workspace_root,
spawner_cwd,
source_workdir_session,
registry,
spawner_manifest,
spawner_scope,
prompts,
)
}
@@ -969,10 +904,9 @@ fn sub_worker_spawn_tool_impl(
parent_notifications: ParentNotificationTarget,
runtime_base: PathBuf,
workspace_root: PathBuf,
spawner_cwd: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>,
registry: Arc<SpawnedWorkerRegistry>,
spawner_manifest: WorkerManifest,
spawner_scope: SharedScope,
prompts: Arc<ArcSwap<PromptCatalog>>,
) -> ToolDefinition {
Arc::new(move || {
@@ -1001,14 +935,11 @@ fn sub_worker_spawn_tool_impl(
parent_notifications.clone(),
runtime_base.clone(),
workspace_root.clone(),
spawner_cwd.clone(),
source_workdir_session.clone(),
registry.clone(),
spawner_manifest.clone(),
prompts.load_full().source(),
available_profiles,
spawner_scope.clone(),
DelegationScope::from_config(&spawner_manifest.delegation_scope)
.expect("resolved Worker manifest has a valid delegation scope"),
));
(meta, tool)
})
@@ -1017,6 +948,7 @@ fn sub_worker_spawn_tool_impl(
#[cfg(test)]
mod tests {
use super::*;
use manifest::{DelegationScope, Permission, Scope, SharedScope};
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::Duration;
@@ -1035,25 +967,63 @@ mod tests {
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceResponse,
};
#[test]
fn missing_active_workdir_session_fails_deterministically() {
let error = require_active_workdir_session(None).unwrap_err();
assert!(matches!(
error,
ToolError::InvalidArgument(message)
if message.contains("requires an active Workdir session")
));
}
#[test]
fn workdir_scope_uses_logical_relative_paths() {
let rules = parse_workdir_scope(&[
ScopeRuleInput {
target: ".".to_string(),
permission: PermissionInput::Read,
recursive: true,
},
ScopeRuleInput {
target: "src".to_string(),
permission: PermissionInput::Write,
recursive: false,
},
])
.unwrap();
assert_eq!(rules[0].target.as_str(), "");
assert_eq!(rules[1].target.as_str(), "src");
for target in ["/host/path", "../escape"] {
let error = parse_workdir_scope(&[ScopeRuleInput {
target: target.to_string(),
permission: PermissionInput::Read,
recursive: true,
}])
.unwrap_err();
assert!(matches!(error, ToolError::InvalidArgument(_)));
}
}
#[test]
fn reviewer_handoff_requires_explicit_builtin_profile_and_read_only_scope() {
let valid: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
"name":"reviewer","task":"review","profile":"builtin:reviewer",
"scope":[{"target":"/tmp/work","permission":"read"}],
"scope":[{"target":"work","permission":"read"}],
"review":{"ticket_id":"T1"}
}))
.unwrap();
assert!(validate_reviewer_handoff(&valid).is_ok());
let wrong_profile: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
"name":"reviewer","task":"review","profile":"builtin:coder",
"scope":[{"target":"/tmp/work","permission":"read"}],
"scope":[{"target":"work","permission":"read"}],
"review":{"ticket_id":"T1"}
}))
.unwrap();
assert!(validate_reviewer_handoff(&wrong_profile).is_err());
let writable: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
"name":"reviewer","task":"review","profile":"builtin:reviewer",
"scope":[{"target":"/tmp/work","permission":"write"}],
"scope":[{"target":"work","permission":"write"}],
"review":{"ticket_id":"T1"}
}))
.unwrap();
@@ -1132,19 +1102,26 @@ extract_threshold = 4000
let fail_requests = Arc::new(AtomicBool::new(false));
let prompt_loader = PromptCatalogSource::builtins_only();
let (parent_method_tx, mut parent_method_rx) = mpsc::channel(8);
let source_workdir_session = workdir::delegation_capable_session(Arc::new(
workdir::LocalWorkdirSession::materialized_bound(
workdir::Workdir::new("test-workdir"),
workspace_root.clone(),
workspace_root.clone(),
spawner_scope.clone(),
workdir::WorkdirSessionCapabilities::ALL,
),
));
let tool = SubWorkerSpawnTool::new(
"parent".into(),
workspace_context,
ParentNotificationTarget::Controller(parent_method_tx.downgrade()),
runtime.path().to_path_buf(),
workspace_root.clone(),
workspace_root.clone(),
Some(source_workdir_session),
registry.clone(),
manifest.clone(),
prompt_loader,
available_profiles,
spawner_scope.clone(),
DelegationScope::from_config(&manifest.delegation_scope).unwrap(),
)
.with_internal_client(Box::new(ScriptedInternalClient {
calls: calls.clone(),
@@ -1160,8 +1137,8 @@ extract_threshold = 4000
"instruction": "role.reviewer",
"task": "review immutable commit",
"scope": [{
"target": workspace_root.clone(),
"permission": "write",
"target": ".",
"permission": "read",
"recursive": true
}]
});
@@ -1188,16 +1165,30 @@ extract_threshold = 4000
.await
.expect("spawn project reviewer as Internal Worker");
assert!(output.summary.contains("internal worker `reviewer-child`"));
assert!(!spawner_scope.snapshot().is_writable(&workspace_root));
assert!(spawner_scope.snapshot().is_writable(&workspace_root));
let record = registry
.get_internal("reviewer-child")
.expect("Internal reviewer registry record");
assert!(record.installed_tools.iter().any(|name| name == "Read"));
for denied in ["Write", "Edit", "Bash"] {
assert!(
!record.installed_tools.iter().any(|name| name == denied),
"read-only child unexpectedly received {denied}: {:?}",
record.installed_tools
);
}
assert!(
!record
.installed_tools
.iter()
.any(|name| matches!(name.as_str(), "WorkdirAttachSelf" | "WorkdirDetachSelf"))
);
assert_eq!(
record.session.wait_until_idle().await,
crate::internal_worker::InternalWorkerSessionStatus::Idle
);
assert_eq!(calls.load(Ordering::SeqCst), 1);
assert!(observed_parent_write_revoked.load(Ordering::SeqCst));
assert!(!observed_parent_write_revoked.load(Ordering::SeqCst));
assert!(observed_instruction_override.load(Ordering::SeqCst));
let completion = tokio::time::timeout(Duration::from_secs(1), parent_method_rx.recv())
.await
@@ -1295,7 +1286,11 @@ extract_threshold = 4000
assert_eq!(calls.load(Ordering::SeqCst), 3);
assert!(
spawner_scope.snapshot().is_writable(&workspace_root),
"Failed terminal child must automatically reclaim its delegated write scope"
"Failed terminal child must release its delegated Workdir session"
);
assert!(
!record.workdir_delegation.is_active(),
"failed child must revoke cloned scoped sessions"
);
assert!(registry.get_internal("reviewer-child").is_some());
@@ -1315,7 +1310,7 @@ extract_threshold = 4000
)
.await
.unwrap();
assert!(!spawner_scope.snapshot().is_writable(&workspace_root));
assert!(spawner_scope.snapshot().is_writable(&workspace_root));
drop(list);
drop(send);
drop(stop);
@@ -1343,45 +1338,6 @@ extract_threshold = 4000
);
}
#[test]
fn spawn_worker_validate_cwd_requires_absolute_existing_directory_in_child_scope() {
let root = TempDir::new().unwrap();
let child_cwd = root.path().join("child");
std::fs::create_dir(&child_cwd).unwrap();
let file_path = root.path().join("file.txt");
std::fs::write(&file_path, "not a dir").unwrap();
let outside = TempDir::new().unwrap();
let missing = root.path().join("missing");
let rules = vec![abs_rule(root.path(), Permission::Write)];
assert_eq!(
validate_spawn_cwd(None, &rules, root.path()).unwrap(),
root.path()
);
assert_eq!(
validate_spawn_cwd(Some(&child_cwd), &rules, root.path()).unwrap(),
std::fs::canonicalize(&child_cwd).unwrap()
);
for (cwd, expected) in [
(Path::new("relative"), "must be absolute"),
(missing.as_path(), "does not exist"),
(file_path.as_path(), "must be a directory"),
(
outside.path(),
"outside the child's delegated readable scope",
),
] {
let err = validate_spawn_cwd(Some(cwd), &rules, root.path()).unwrap_err();
match err {
ToolError::InvalidArgument(message) => {
assert!(message.contains(expected), "{message}")
}
other => panic!("expected InvalidArgument, got {other:?}"),
}
}
}
#[test]
fn orchestration_delegation_allows_root_read_and_worktree_writes_not_root_writes() {
let tmp = TempDir::new().unwrap();
+77 -5
View File
@@ -49,7 +49,8 @@ use workdir::workspace::{
WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse,
WorkingDirectoryDiagnostic, WorkingDirectoryDiagnosticSeverity,
WorkingDirectoryListResponse as BrowserWorkingDirectoryListResponse, WorkingDirectoryOccupancy,
WorkingDirectoryStatusKind, WorkingDirectorySummary,
WorkingDirectoryStatusKind, WorkingDirectorySummary, WorkspaceWorkdirSessionFence,
WorkspaceWorkdirSessionOperationRequest,
};
use worker::feature::builtin::{WorkerObservationSubject, WorkerObservationSubjectRef};
use worker_runtime::resource::{BackendResourceError, BackendResourceFetchRequest};
@@ -1523,6 +1524,10 @@ pub fn build_router(api: WorkspaceApi) -> Router {
post(scoped_attach_current_worker_workdir)
.delete(scoped_detach_current_worker_workdir),
)
.route(
"/api/w/{workspace_id}/workers/self/workdir-session/fence",
get(scoped_current_worker_workdir_session_fence),
)
.route(
"/api/w/{workspace_id}/workers/self/workdir-session/operations",
post(scoped_execute_current_worker_workdir_operation),
@@ -5043,7 +5048,7 @@ async fn scoped_attach_current_worker_workdir(
worker: worker.clone(),
workdir_id: workdir_id.to_string(),
role: "attachment".to_string(),
linked_at: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
linked_at: Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
unlinked_at: None,
})?;
if let Err(error) = open_current_worker_workdir_session_locked(&api, &worker, &link).await {
@@ -5086,19 +5091,62 @@ async fn scoped_detach_current_worker_workdir(
}))
}
async fn scoped_current_worker_workdir_session_fence(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
headers: HeaderMap,
) -> ApiResult<Json<WorkspaceWorkdirSessionFence>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let worker = current_worker_identity(&api, &path.workspace_id, &headers)?;
let session_lock = current_worker_session_lock(&api, &worker);
let _session_guard = session_lock.lock().await;
let link = current_worker_active_attachment(&api, &worker)?;
Ok(Json(WorkspaceWorkdirSessionFence {
value: current_worker_workdir_session_fence(&link),
}))
}
fn current_worker_workdir_session_fence(link: &WorkerWorkdirLinkRecord) -> String {
format!("v1:{}\0{}", link.workdir_id, link.linked_at)
}
fn validate_current_worker_workdir_session_fence(
link: &WorkerWorkdirLinkRecord,
expected: Option<&str>,
) -> Result<()> {
if expected.is_some_and(|expected| expected != current_worker_workdir_session_fence(link)) {
Err(Error::WorkdirAttachmentConflict(
"delegated Workdir session attachment changed".to_string(),
))
} else {
Ok(())
}
}
async fn scoped_execute_current_worker_workdir_operation(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
headers: HeaderMap,
Json(operation): Json<WorkdirSessionOperation>,
Json(request): Json<WorkspaceWorkdirSessionOperationRequest>,
) -> ApiResult<Json<WorkdirSessionOperationResult>> {
validate_workspace_scope(&api, &path.workspace_id)?;
let worker = current_worker_identity(&api, &path.workspace_id, &headers)?;
let session_lock = current_worker_session_lock(&api, &worker);
let _session_guard = session_lock.lock().await;
let link = current_worker_active_attachment(&api, &worker)?;
let session = open_current_worker_workdir_session_locked(&api, &worker, &link).await?;
let result = execute_workdir_session_operation(&session, operation)
validate_current_worker_workdir_session_fence(
&link,
request.expected_session_fence.as_deref(),
)?;
let source = open_current_worker_workdir_session_locked(&api, &worker, &link).await?;
let applied = workdir::apply_delegation_chain(source, request.delegations)
.await
.map_err(|error| Error::RuntimeOperationFailed {
runtime_id: worker.runtime_id.clone(),
code: "workdir_session_delegation_failed".to_string(),
message: error.to_string(),
})?;
let result = execute_workdir_session_operation(&applied.scoped_session, request.operation)
.await
.map_err(|error| Error::RuntimeOperationFailed {
runtime_id: worker.runtime_id.clone(),
@@ -16167,6 +16215,30 @@ mod tests {
assert_eq!(response.status(), StatusCode::BAD_REQUEST);
}
#[test]
fn delegated_workdir_session_fence_rejects_reattached_link() {
let first = WorkerWorkdirLinkRecord {
workspace_id: "workspace-a".to_string(),
worker: workdir::workspace::RuntimeWorkerRef::new("runtime-a", "worker-a"),
workdir_id: "workdir-a".to_string(),
role: "primary".to_string(),
linked_at: "2026-01-01T00:00:00Z".to_string(),
unlinked_at: None,
};
let expected = current_worker_workdir_session_fence(&first);
assert!(validate_current_worker_workdir_session_fence(&first, None).is_ok());
assert!(validate_current_worker_workdir_session_fence(&first, Some(&expected)).is_ok());
let reattached = WorkerWorkdirLinkRecord {
linked_at: "2026-01-01T00:00:01Z".to_string(),
..first
};
assert!(matches!(
validate_current_worker_workdir_session_fence(&reattached, Some(&expected)),
Err(Error::WorkdirAttachmentConflict(_))
));
}
#[tokio::test]
async fn backend_workdir_session_proxy_executes_typed_operations() {
use manifest::Scope;
+7 -1
View File
@@ -8,7 +8,13 @@ The assigned Coder owns its review/fix loop and launches Reviewer SubWorkers its
Treat the current linked Merge Request as implementation-completion authority. A current provider-resolved source ref, commit/repository evidence, an effective approval for that exact subject, review freshness after the latest substantive Ticket item edit, and no unresolved request-changes are sufficient; do not require an `implementation_report`. Human summaries remain optional audit context. Recheck `ShowTicket` and `MergeRequestReadinessCheck` immediately before guarded integration. Require the Merge Request source selector to remain on the exact reviewed commit; any source movement requires a fresh Reviewer attempt.
Before integration, run `MergeRequestReadinessCheck` and reread the Ticket, current assignment, and exact approved subject. In the Orchestrator Workdir, use the Ticket repository `origin` transport to fetch the current target selector and immutable source selector, verify both against readiness evidence, apply the selected fast-forward or merge strategy, and validate the resulting tree. Push only a result that descends from the observed target, using a guarded non-force push whose expected old target is `target_ref_before`; reject target movement and conflicts rather than rewriting the remote. Verify the remote target now resolves exactly to `target_ref_after`, then call `MergeRequestComplete` with that before/after evidence and the authoritative approval event. Never mutate a Server-side repository path or use local `git update-ref` as integration authority.
Before integration, run `MergeRequestReadinessCheck` and reread the Ticket, current assignment, and exact approved subject. Treat the provider-resolved `selector_from` as the `merge_from` branch and `selector_to` as the `merge_to` branch. Obtain their exact approved source hash and `target_ref_before`, selected merge strategy, and approval event from authoritative Merge Request evidence.
Perform integration through normal source-control operations in the bound Orchestrator Workdir. Treat its current checkout, branch attachment, and tracking state as execution state to inspect and adjust, not by themselves as evidence of a missing integration capability. Ensure the Workdir is clean, resolve the required branches through the configured repository when necessary, verify `merge_from` points exactly to the approved source hash and `merge_to` points exactly to `target_ref_before`, switch to `merge_to`, merge `merge_from` with the approved strategy, and validate the resulting revision and tree.
Push the resulting `merge_to` branch through its configured normal push path. Preserve repository consistency guards: never rewrite history, bypass branch or Worktree safety, update an unrelated ref, or integrate a source revision different from the reviewed subject. Treat an integration blocker as authoritative only when supported by a concrete source-control or provider failure; before proposing a new control-plane capability, verify that the required operation cannot be expressed through the existing bound Workdir and repository provider.
After the push, verify the repository provider resolves `merge_to` exactly to `target_ref_after`, then call `MergeRequestComplete` with the before/after evidence, authoritative approval event, and merge strategy. `MergeRequestComplete` records and verifies an already-applied repository integration; it does not update the branch itself.
If the repository push succeeds but completion recording fails, do not push again or invent a new result. Retry the same completion operation and evidence: while no completion event exists, the Server requires the target to remain at the exact `target_ref_after` before it records `MergeResult`, moves the Ticket to `done`, and closes the current assignment atomically. Once that exact operation is recorded, later target movement does not invalidate an idempotent replay of the recorded result. Before recording, any other observed target is a stale/conflicting completion and must fail closed.
+12 -1
View File
@@ -53,6 +53,12 @@ export type InFlightBlock = { "kind": "text", text: string, finished?: boolean,
export type InFlightSnapshot = { blocks?: Array<InFlightBlock>, };
export type InternalWorkerKind = "sub_worker";
export type InternalWorkerRef = { session_id: string, name: string, parent_session_id?: string | null, kind: InternalWorkerKind, };
export type InternalWorkerSnapshot = { worker: InternalWorkerRef, revision: number, entries: Array<unknown>, status: WorkerStatus, error?: string | null, in_flight?: InFlightSnapshot, internal_workers?: Array<InternalWorkerSnapshot>, };
export type Greeting = { worker_name: string, cwd: string, provider: string, model: string, scope_summary: string, tools: Array<string>,
/**
* Model context window in tokens. Always filled by the Worker greeting.
@@ -167,4 +173,9 @@ output?: string | null, is_error: boolean, } } | { "event": "usage", "data": { i
* Unfinished model output that has already streamed in the current
* run but is not yet represented by committed snapshot entries.
*/
in_flight?: InFlightSnapshot, } } | { "event": "segment_rotated", "data": { entry: unknown, } } | { "event": "status", "data": { status: WorkerStatus, } } | { "event": "completions", "data": { kind: CompletionKind, entries: Array<CompletionEntry>, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array<RewindTarget>, } } | { "event": "rewind_applied", "data": { entries: Array<unknown>, input: Array<Segment>, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compact_start" } | { "event": "compact_done", "data": { new_segment_id: string, } } | { "event": "compact_failed", "data": { error: string, } } | { "event": "shutdown" };
in_flight?: InFlightSnapshot,
/**
* Parent-owned Internal Worker sessions visible to this client.
* Service-private Internal Workers are deliberately excluded.
*/
internal_workers?: Array<InternalWorkerSnapshot>, } } | { "event": "internal_worker", "data": { worker: InternalWorkerRef, revision: number, event: Event, } } | { "event": "segment_rotated", "data": { entry: unknown, } } | { "event": "status", "data": { status: WorkerStatus, } } | { "event": "completions", "data": { kind: CompletionKind, entries: Array<CompletionEntry>, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array<RewindTarget>, } } | { "event": "rewind_applied", "data": { entries: Array<unknown>, input: Array<Segment>, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compact_start" } | { "event": "compact_done", "data": { new_segment_id: string, } } | { "event": "compact_failed", "data": { error: string, } } | { "event": "shutdown" };
@@ -1266,6 +1266,82 @@ Deno.test("projectConsole mirrors live TaskCreate and TaskUpdate calls", () => {
}]);
});
Deno.test("Internal Worker output stays separate and revision-fenced", () => {
const worker = {
session_id: "child-session",
name: "research",
parent_session_id: "parent-session",
kind: "sub_worker" as const,
};
const projector = createConsoleProjector();
let projection = projector.append([{
eventId: "1",
event: {
event: "internal_worker",
data: {
worker,
revision: 2,
event: { event: "text_done", data: { text: "child output" } },
},
},
}]);
assertEquals(projection.lines, []);
assertEquals(projection.internalWorkers.length, 1);
assertEquals(projection.internalWorkers[0].console.lines[0].body, "child output");
projection = projector.append([{
eventId: "2",
event: {
event: "internal_worker",
data: {
worker,
revision: 1,
event: { event: "text_done", data: { text: "stale" } },
},
},
}]);
assertEquals(projection.internalWorkers[0].console.lines.length, 1);
});
Deno.test("parent snapshot authoritatively replaces Internal Worker projections", () => {
const event = snapshotEvent("/repo");
if (event.event !== "snapshot") throw new Error("snapshot fixture expected");
event.data.internal_workers = [{
worker: {
session_id: "replacement",
name: "replacement",
parent_session_id: "parent-session",
kind: "sub_worker",
},
revision: 4,
entries: [],
status: "idle",
in_flight: { blocks: [] },
internal_workers: [],
}];
const projector = createConsoleProjector();
projector.append([{
eventId: "old",
event: {
event: "internal_worker",
data: {
worker: {
session_id: "old",
name: "old",
parent_session_id: "parent-session",
kind: "sub_worker",
},
revision: 1,
event: { event: "status", data: { status: "running" } },
},
},
}]);
const projection = projector.append([{ eventId: "snapshot", event }]);
assertEquals(projection.internalWorkers.map((worker) => worker.worker.session_id), [
"replacement",
]);
});
Deno.test("snapshot restores TaskStore state from system history", () => {
const taskSnapshot =
`[Session TaskStore snapshot]\n\n\`\`\`json\n{\n "tasks": [{"taskid": 3, "status": "pending", "subject": "Restored", "description": "From compaction"}]\n}\n\`\`\``;
@@ -3,6 +3,8 @@ import type {
Event as ProtocolEvent,
InFlightBlock,
InFlightToolCallState,
InternalWorkerRef,
InternalWorkerSnapshot,
Segment,
} from "$lib/generated/protocol";
import { workspaceRoute } from "$lib/workspace/api/http";
@@ -63,6 +65,26 @@ export type ConsoleLine = {
toolCall?: ToolCallView;
};
export type InternalWorkerProjection = {
worker: InternalWorkerRef;
revision: number;
console: ConsoleProjection;
};
export type FlattenedInternalWorkerProjection = InternalWorkerProjection & {
depth: number;
};
export function flattenInternalWorkers(
workers: InternalWorkerProjection[],
depth = 0,
): FlattenedInternalWorkerProjection[] {
return workers.flatMap((worker) => [
{ ...worker, depth },
...flattenInternalWorkers(worker.console.internalWorkers, depth + 1),
]);
}
export type ConsoleProjection = {
lines: ConsoleLine[];
tasks: ConsoleTask[];
@@ -71,6 +93,7 @@ export type ConsoleProjection = {
usage: string | null;
cwd: string | null;
lastEventId: string | null;
internalWorkers: InternalWorkerProjection[];
};
export type ConsoleTimelineLineSelection = {
@@ -155,6 +178,7 @@ export function emptyConsoleProjection(): ConsoleProjection {
usage: null,
cwd: null,
lastEventId: null,
internalWorkers: [],
};
}
@@ -193,9 +217,50 @@ function projectVisibleConsole(
return {
...projection,
lines: aggregateReadToolLines(projection.lines),
internalWorkers: projection.internalWorkers.map((worker) => ({
...worker,
console: projectVisibleConsole(worker.console),
})),
};
}
function projectInternalWorkerSnapshot(
snapshot: InternalWorkerSnapshot,
eventId: string,
cwd: string | null,
): InternalWorkerProjection {
const console = snapshotProjectionFromEntries(
`${eventId}:internal:${snapshot.worker.session_id}:snapshot`,
snapshot.entries,
cwd,
);
console.status = snapshot.status;
for (const block of snapshot.in_flight?.blocks ?? []) {
console.lines.push(
inFlightLine(
`${eventId}:internal:${snapshot.worker.session_id}:in-flight`,
block,
cwd,
),
);
}
if (snapshot.error) {
console.lines.push({
id: `${eventId}:internal:${snapshot.worker.session_id}:error`,
kind: "error",
title: "Error",
body: snapshot.error,
eventId,
source: "event",
error: true,
});
}
console.internalWorkers = (snapshot.internal_workers ?? []).map((child) =>
projectInternalWorkerSnapshot(child, eventId, cwd)
);
return { worker: snapshot.worker, revision: snapshot.revision, console };
}
export function applyProtocolEvent(
projection: ConsoleProjection,
envelope: { eventId: string; event: ProtocolEvent },
@@ -208,6 +273,7 @@ export function applyProtocolEvent(
usage: projection.usage,
cwd: projection.cwd,
lastEventId: envelope.eventId,
internalWorkers: [...projection.internalWorkers],
};
const event = envelope.event;
@@ -322,6 +388,34 @@ export function applyProtocolEvent(
for (const block of event.data.in_flight?.blocks ?? []) {
next.lines.push(inFlightLine(envelope.eventId, block, next.cwd));
}
next.internalWorkers = (event.data.internal_workers ?? []).map((worker) =>
projectInternalWorkerSnapshot(worker, envelope.eventId, next.cwd)
);
break;
}
case "internal_worker": {
const existingIndex = next.internalWorkers.findIndex((worker) =>
worker.worker.session_id === event.data.worker.session_id
);
const existing = existingIndex >= 0
? next.internalWorkers[existingIndex]
: {
worker: event.data.worker,
revision: 0,
console: emptyConsoleProjection(),
};
if (event.data.revision <= existing.revision) break;
const updated: InternalWorkerProjection = {
worker: event.data.worker,
revision: event.data.revision,
console: applyProtocolEvent(existing.console, {
eventId:
`${envelope.eventId}:internal:${event.data.worker.session_id}:${event.data.revision}`,
event: event.data.event,
}),
};
if (existingIndex >= 0) next.internalWorkers[existingIndex] = updated;
else next.internalWorkers.push(updated);
break;
}
case "status":
@@ -1184,6 +1278,7 @@ function snapshotProjectionFromEntries(
usage: null,
cwd,
lastEventId: eventId,
internalWorkers: [],
};
entries.forEach((entry, index) =>
applyLogEntry(projection, `${eventId}-snapshot-${index}`, entry)
@@ -18,6 +18,7 @@
import { fitTextarea } from "$lib/workspace/console/textarea-fit";
import {
createConsoleProjector,
flattenInternalWorkers,
isConsoleProjectionEvent,
selectConsoleTimelineLines,
type ConsoleEventInput,
@@ -152,6 +153,9 @@
const lines = $derived(consoleProjection.lines);
const tasks = $derived(consoleProjection.tasks);
const internalWorkers = $derived(
flattenInternalWorkers(consoleProjection.internalWorkers),
);
const timelineLayout = $derived(
buildTimelineLayout(lines, eventObservedAtVersion, consoleScroll),
);
@@ -1245,6 +1249,28 @@
</ol>
{/if}
</article>
{#each internalWorkers as internal (internal.worker.session_id)}
<section
class="card internal-worker-pane"
style={`--internal-worker-depth: ${internal.depth}`}
aria-label={`SubWorker ${internal.worker.name}`}
>
<header class="internal-worker-header">
<strong>{internal.worker.name}</strong>
<span>{internal.console.status ?? "unknown"}</span>
</header>
{#if internal.console.lines.length === 0}
<p>No output yet.</p>
{:else}
<ol class="console-log">
{#each internal.console.lines as item (item.id)}
<ConsoleLineItem {item} />
{/each}
</ol>
{/if}
</section>
{/each}
</div>
<ConsoleTimeline
@@ -1771,6 +1797,23 @@
margin-right: auto;
}
.internal-worker-pane {
margin: 0.75rem 0 0 calc((var(--internal-worker-depth) + 1) * 1rem);
border-left: 3px solid var(--color-border-strong, currentColor);
}
.internal-worker-header {
display: flex;
justify-content: space-between;
gap: 1rem;
margin-bottom: 0.5rem;
font-family: var(--font-mono);
}
.internal-worker-header span {
color: var(--color-text-muted);
}
@media (max-width: 960px) {
.console-history.with-task-pane {
grid-template-columns: minmax(0, 1fr);