Merge remote-tracking branch 'origin/develop' into ticket/t-768-subject-grants

# Conflicts:
#	crates/worker-runtime/src/worker_backend.rs
#	openapi/server-api.json
#	web/workspace/src/lib/generated/companion-api.ts
#	web/workspace/src/lib/generated/repository-api.ts
#	web/workspace/src/lib/generated/runtime-api.ts
#	web/workspace/src/lib/generated/ticket-api.ts
#	web/workspace/src/lib/generated/worker-launch-api.ts
This commit is contained in:
2026-10-20 06:24:11 +09:00
116 changed files with 9151 additions and 2358 deletions
+1
View File
@@ -56,6 +56,7 @@ server-api = { path = "../server-api" }
workdir.workspace = true
[dev-dependencies]
workdir = { workspace = true, features = ["http-client"] }
futures.workspace = true
agen.workspace = true
serial_test = "3.4.0"
+5 -8
View File
@@ -67,7 +67,7 @@ use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use tokio::net::TcpListener;
use workdir::{
CommandOutput, CommandStatus, WorkdirSessionHandle, dispatch_workdir_session_operation,
CommandOutput, WorkdirSessionHandle, dispatch_workdir_session_operation,
http::{
OpenWorkdirSessionRequest, OpenWorkdirSessionResponse, WorkdirSessionId,
WorkdirSessionOperation, WorkdirSessionOperationRequest, WorkdirSessionOperationResult,
@@ -914,13 +914,9 @@ async fn run_workdir_session_operation(
{
Ok(result) => result?,
Err(_) => CommandOutput {
exit_code: None,
timed_out: false,
status: CommandStatus::Running,
content: String::new(),
next_cursor: Some(cursor),
truncated: false,
output_path: None,
chunks: Vec::new(),
next_cursor: cursor,
terminal: None,
},
};
WorkdirSessionOperationResult::CommandOutput(output)
@@ -2426,6 +2422,7 @@ pub enum RuntimeHttpServerError {
#[cfg(test)]
mod tests {
mod command_transport_tests;
#[cfg(feature = "fs-store")]
mod session_material_tests;
@@ -0,0 +1,408 @@
//! Loopback integration: production RemoteWorkdirSession, signed HTTP middleware,
//! command operation handler, local process pipes, and JSON DTOs. Only opening the
//! pre-seeded session is a fixture; no Runtime Worker/materializer is claimed here.
use super::*;
use std::time::Duration;
use workdir::http::{
RemoteWorkdirSession, WorkdirHttpAuthorization, WorkdirHttpAuthorizationFuture,
};
use workdir::{CommandOutputRequest, CommandRequest, CommandStream, WorkdirSession};
struct SignedAuthorization {
identity: RuntimeIdentityMaterial,
trust: WorkspaceIssuerTrustRecord,
}
impl std::fmt::Debug for SignedAuthorization {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str("SignedAuthorization")
}
}
impl WorkdirHttpAuthorization for SignedAuthorization {
fn bearer_token(
&self,
method: String,
path: String,
body: Vec<u8>,
) -> WorkdirHttpAuthorizationFuture {
let claims = signed_claims(&self.trust, &method, &path, &body);
let token =
issue_workspace_capability_token(&self.identity.signing_key().unwrap(), &claims)
.unwrap();
Box::pin(async move { Ok(token) })
}
}
struct Harness {
server: tokio::task::JoinHandle<()>,
session: WorkdirSessionHandle,
_root: tempfile::TempDir,
remote: RemoteWorkdirSession,
client: reqwest::Client,
operation_url: url::Url,
authorization: Arc<SignedAuthorization>,
}
impl Drop for Harness {
fn drop(&mut self) {
self.server.abort();
}
}
impl Harness {
async fn raw_operation(&self, operation: serde_json::Value) -> reqwest::Response {
let body = serde_json::to_vec(&serde_json::json!({"operation": operation})).unwrap();
let token = self
.authorization
.bearer_token(
"POST".into(),
self.operation_url.path().into(),
body.clone(),
)
.await
.unwrap();
self.client
.post(self.operation_url.clone())
.bearer_auth(token)
.header("content-type", "application/json")
.body(body)
.send()
.await
.unwrap()
}
async fn new() -> Self {
let root = tempfile::tempdir().unwrap();
let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound(
Workdir::new("wd-transport"),
root.path().to_path_buf(),
root.path().to_path_buf(),
SharedScope::new(Scope::writable(root.path()).unwrap()),
WorkdirSessionCapabilities::ALL,
));
let identity = RuntimeIdentityMaterial::generate("transport-key").unwrap();
let trust = workspace_trust(&identity);
let state = RuntimeHttpState {
runtime: Runtime::new_memory(),
local_token: None,
workspace_auth: Some(Arc::new(workspace_test_auth(vec![trust.clone()]))),
workdir_sessions: Arc::new(Mutex::new(HashMap::from([(
"session-transport".into(),
RuntimeHttpWorkdirSession {
owner: RuntimeWorkspaceScope::new("workspace-a", "https://backend.test"),
session: session.clone(),
},
)]))),
ssh_keyscan_program: Arc::new(PathBuf::from("ssh-keyscan")),
};
let app = Router::new()
.route(
"/v1/working-directories/{id}/sessions",
post(|| async {
Json(OpenWorkdirSessionResponse {
session_id: WorkdirSessionId::new("session-transport").unwrap(),
workdir_id: Workdir::new("wd-transport").id().clone(),
capabilities: WorkdirSessionCapabilities::ALL,
})
}),
)
.route(
"/v1/workdir-sessions/{session_id}/operations",
post(run_workdir_session_operation),
)
.route(
"/v1/workdir-sessions/{session_id}",
delete(close_workdir_session),
)
.with_state(state.clone())
.layer(middleware::from_fn_with_state(state, require_runtime_auth));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base: url::Url = format!("http://{}", listener.local_addr().unwrap())
.parse()
.unwrap();
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let mut headers = reqwest::header::HeaderMap::new();
headers.insert(
RUNTIME_WORKSPACE_SCOPE_HEADER,
"workspace-a".parse().unwrap(),
);
let client = reqwest::Client::builder()
.no_proxy()
.default_headers(headers)
.build()
.unwrap();
let authorization = Arc::new(SignedAuthorization { identity, trust });
let operation_url = base
.join("/v1/workdir-sessions/session-transport/operations")
.unwrap();
let remote = RemoteWorkdirSession::open_with_authorization(
client.clone(),
base,
authorization.clone(),
Workdir::new("wd-transport").id().clone(),
OpenWorkdirSessionRequest::default(),
)
.await
.unwrap();
Self {
server,
session,
_root: root,
remote,
client,
operation_url,
authorization,
}
}
}
#[tokio::test]
async fn real_http_command_pages_preserve_binary_streams_replay_and_nonzero_exit() {
tokio::time::timeout(Duration::from_secs(20), async {
let harness = Harness::new().await;
let handle = harness.remote.start_command(CommandRequest {
transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES,
command: "head -c 2097152 /dev/zero | tr '\\0' x; printf '\\342\\202\\254\\377\\000'; printf 'stderr\\346\\227\\245' >&2; exit 9".into(),
timeout_secs: 15, cwd: WorkdirPath::root(), tool_call_id: Some("http-tool-call".into()),
}).await.unwrap();
let mut cursor = 0;
let mut stdout = Vec::new();
let mut stderr = Vec::new();
let terminal = loop {
let request = CommandOutputRequest { handle: handle.clone(), cursor, limit: 16384, wait: true };
let page = harness.remote.command_output(request.clone()).await.unwrap();
assert_eq!(harness.remote.command_output(request).await.unwrap(), page);
assert!(page.chunks.iter().map(|chunk| chunk.bytes.len()).sum::<usize>() <= 16384);
for chunk in page.chunks {
assert_eq!(chunk.sequence, cursor);
cursor += chunk.bytes.len() as u64;
let target = match chunk.stream { CommandStream::Stdout => &mut stdout, CommandStream::Stderr => &mut stderr };
assert_eq!(chunk.offset, target.len() as u64);
target.extend(chunk.bytes);
}
assert_eq!(cursor, page.next_cursor);
if let Some(terminal) = page.terminal { break terminal; }
};
let mut expected = vec![b'x'; 2097152];
expected.extend_from_slice(b"\xe2\x82\xac\xff\0");
assert_eq!(stdout, expected);
assert_eq!(stderr, b"stderr\xe6\x97\xa5");
assert_eq!(terminal.status, workdir::CommandStatus::Failed);
assert_eq!(terminal.exit_code, Some(9));
assert!(harness.remote.command_output(CommandOutputRequest { handle, cursor: 0, limit: 1, wait: false }).await.is_err());
harness.remote.close().await.unwrap();
assert!(harness.session.command_snapshot().is_empty());
}).await.expect("bounded HTTP command transport");
}
#[tokio::test]
async fn real_http_cancel_drains_tail_before_terminal_and_timeout_is_distinct() {
tokio::time::timeout(Duration::from_secs(15), async {
let harness = Harness::new().await;
for (command, timeout, cancel, expected) in [
(
"printf ready; exec yes x",
10,
true,
workdir::CommandStatus::Cancelled,
),
(
"printf ready; exec tail -f /dev/null",
1,
false,
workdir::CommandStatus::TimedOut,
),
] {
let handle = harness
.remote
.start_command(CommandRequest {
transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES,
command: command.into(),
timeout_secs: timeout,
cwd: WorkdirPath::root(),
tool_call_id: None,
})
.await
.unwrap();
let mut cursor = 0;
let first = harness
.remote
.command_output(CommandOutputRequest {
handle: handle.clone(),
cursor,
limit: 5,
wait: true,
})
.await
.unwrap();
assert_eq!(
first
.chunks
.iter()
.flat_map(|chunk| chunk.bytes.iter().copied())
.collect::<Vec<_>>(),
b"ready"
);
cursor = first.next_cursor;
if cancel {
harness.remote.cancel_command(handle.clone()).await.unwrap();
}
loop {
let page = harness
.remote
.command_output(CommandOutputRequest {
handle: handle.clone(),
cursor,
limit: workdir::MAX_COMMAND_READ_BYTES,
wait: true,
})
.await
.unwrap();
cursor = page.next_cursor;
if let Some(terminal) = page.terminal {
assert_eq!(terminal.status, expected);
break;
}
}
}
harness.remote.close().await.unwrap();
})
.await
.expect("bounded cancellation and timeout over HTTP");
}
#[tokio::test]
async fn real_http_command_budget_bounds_unique_payload_and_reports_larger_seen_totals() {
tokio::time::timeout(Duration::from_secs(20), async {
let harness = Harness::new().await;
for budget in [0, 12345] {
let handle = harness
.remote
.start_command(CommandRequest {
command: "head -c 1048576 /dev/zero; head -c 65536 /dev/zero >&2; exit 9"
.into(),
timeout_secs: 10,
transport_byte_budget: budget,
cwd: WorkdirPath::root(),
tool_call_id: None,
})
.await
.unwrap();
let mut cursor = 0;
let mut delivered = [0_u64; 2];
let terminal = loop {
let request = CommandOutputRequest {
handle: handle.clone(),
cursor,
limit: 997,
wait: true,
};
let page = harness
.remote
.command_output(request.clone())
.await
.unwrap();
assert_eq!(harness.remote.command_output(request).await.unwrap(), page);
for chunk in page.chunks {
assert_eq!(chunk.sequence, cursor);
cursor += chunk.bytes.len() as u64;
let index = match chunk.stream {
CommandStream::Stdout => 0,
CommandStream::Stderr => 1,
};
assert_eq!(chunk.offset, delivered[index]);
delivered[index] += chunk.bytes.len() as u64;
}
assert_eq!(cursor, page.next_cursor);
assert!(
cursor <= budget,
"RemoteWorkdirSession must not receive bytes beyond budget"
);
if let Some(terminal) = page.terminal {
break terminal;
}
};
assert_eq!(cursor, budget);
assert_eq!(terminal.stdout_bytes, 1048576);
assert_eq!(terminal.stderr_bytes, 65536);
assert_eq!(terminal.stdout_delivered_bytes, delivered[0]);
assert_eq!(terminal.stderr_delivered_bytes, delivered[1]);
assert_eq!(
terminal.truncation_reason,
Some(workdir::CommandOutputTruncation::ByteBudget)
);
assert_eq!(terminal.status, workdir::CommandStatus::Failed);
assert_eq!(terminal.exit_code, Some(9));
assert!(
harness
.remote
.command_output(CommandOutputRequest {
handle: handle.clone(),
cursor: budget + 1,
limit: 1,
wait: false,
})
.await
.is_err()
);
if budget > 0 {
assert!(
harness
.remote
.command_output(CommandOutputRequest {
handle,
cursor: 0,
limit: 1,
wait: false,
})
.await
.is_err()
);
}
}
harness.remote.close().await.unwrap();
})
.await
.expect("bounded HTTP byte-budget transport");
}
#[tokio::test]
async fn real_http_rejects_invalid_or_legacy_budget_requests_before_process_start() {
tokio::time::timeout(Duration::from_secs(10), async {
let harness = Harness::new().await;
let request = CommandRequest {
command: "touch must-not-exist".into(),
timeout_secs: 1,
transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES + 1,
cwd: WorkdirPath::root(),
tool_call_id: None,
};
let oversize =
serde_json::to_value(WorkdirSessionOperation::CommandStart(request)).unwrap();
assert_eq!(
harness.raw_operation(oversize.clone()).await.status(),
reqwest::StatusCode::BAD_REQUEST
);
let mut missing = oversize.clone();
// CommandStart is externally tagged by operation / request.
let object = missing.as_object_mut().unwrap();
let request = object.get_mut("request").unwrap().as_object_mut().unwrap();
request.remove("transport_byte_budget");
assert_eq!(
harness.raw_operation(missing).await.status(),
reqwest::StatusCode::BAD_REQUEST
);
let mut legacy = oversize;
let request = legacy["request"].as_object_mut().unwrap();
request.insert("transport_byte_budget".into(), serde_json::json!(0));
request.insert("spill_dir".into(), serde_json::json!("/private/host"));
assert_eq!(
harness.raw_operation(legacy).await.status(),
reqwest::StatusCode::BAD_REQUEST
);
assert!(!harness._root.path().join("must-not-exist").exists());
harness.remote.close().await.unwrap();
})
.await
.expect("bounded HTTP budget admission");
}
+15 -45
View File
@@ -147,7 +147,6 @@ use worker::{
Worker, WorkerBootstrap, WorkerBootstrapError, WorkerBootstrapLayout,
WorkerControllerTransport, WorkerError, WorkerFilesystemAuthority, WorkerHandle,
WorkerSharedState, WorkerWorkspaceContext, WorkspaceClient, WorkspaceId,
bash_output_dir_for_worker_id,
};
#[cfg(feature = "ws-server")]
@@ -1090,29 +1089,24 @@ fn runtime_local_workdir_session(
root: &Path,
cwd: &Path,
scope: manifest::SharedScope,
command_output_scope: manifest::SharedScope,
capabilities: WorkdirSessionCapabilities,
command_environment: std::collections::BTreeMap<String, String>,
resources: Vec<Arc<dyn workdir::WorkdirSessionResource>>,
) -> WorkdirSessionHandle {
Arc::new(
LocalWorkdirSession::materialized_bound_with_environment_and_command_output_scope(
Workdir::new(workdir_id),
root.to_path_buf(),
cwd.to_path_buf(),
scope,
command_output_scope,
capabilities,
command_environment,
resources,
),
)
Arc::new(LocalWorkdirSession::materialized_bound_with_environment(
Workdir::new(workdir_id),
root.to_path_buf(),
cwd.to_path_buf(),
scope,
capabilities,
command_environment,
resources,
))
}
fn runtime_local_workdir_router(
attachments: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
capabilities: &BTreeMap<WorkdirAttachmentAlias, WorkdirSessionCapabilities>,
command_output_scope: manifest::SharedScope,
authorization_client: Option<Arc<dyn WorkspaceClient>>,
) -> Result<Arc<WorkdirSessionRouter>, String> {
let router = Arc::new(WorkdirSessionRouter::new());
@@ -1127,7 +1121,6 @@ fn runtime_local_workdir_router(
binding.root(),
binding.cwd(),
scope,
command_output_scope.clone(),
capabilities,
binding.command_environment(),
binding
@@ -1153,7 +1146,6 @@ fn restored_workdir_router(
local_attachments: &BTreeMap<WorkdirAttachmentAlias, WorkingDirectoryBinding>,
logical_attachments: &[crate::catalog::LogicalWorkdirAttachment],
claims: &[crate::catalog::WorkingDirectoryAttachmentClaim],
scope: manifest::SharedScope,
workspace_client: Arc<dyn WorkspaceClient>,
) -> Result<Arc<WorkdirSessionRouter>, String> {
let mut local_capabilities = local_attachments
@@ -1173,7 +1165,6 @@ fn restored_workdir_router(
let router = runtime_local_workdir_router(
local_attachments,
&local_capabilities,
scope,
Some(workspace_client.clone()),
)?;
for attachment in logical_attachments {
@@ -1529,7 +1520,6 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
let store = CombinedStore::new(session_store, worker_metadata_store);
let run_dir = self.worker_run_dir(&request.worker_ref)?;
let bash_output_dir = bash_output_dir_for_worker_id(&request.worker_ref.worker_id);
let bootstrap = WorkerBootstrap::new(
manifest,
store,
@@ -1538,7 +1528,6 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
filesystem_authority,
WorkerBootstrapLayout::RuntimeManagedRun {
run_dir: run_dir.clone(),
bash_output_dir,
},
self.controller_transport,
);
@@ -1583,7 +1572,6 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
worker.bind_workdir_sessions(runtime_local_workdir_router(
&request.workdir_attachments,
&workdir_capabilities,
worker.scope().clone(),
Some(worker.workspace_client_handle()),
)?);
if let (Some(runtime_id), Some(workspace_id)) =
@@ -1851,7 +1839,6 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
&request.workdir_attachments,
&request.logical_workdir_attachments,
&request.request.workdir_attachments,
worker.scope().clone(),
worker.workspace_client_handle(),
)?;
worker.bind_workdir_sessions(workdir_sessions);
@@ -1888,13 +1875,11 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
let workspace_client = worker.workspace_client_handle();
let run_dir = self.worker_restore_run_dir(&request.worker_ref, request.operation_id)?;
let bash_output_dir = bash_output_dir_for_worker_id(&request.worker_ref.worker_id);
let cleanup_sessions = worker.workdir_sessions();
let started = PreparedWorker::new(
worker,
WorkerBootstrapLayout::RuntimeManagedRun {
run_dir: run_dir.clone(),
bash_output_dir,
},
self.controller_transport,
)
@@ -3768,7 +3753,6 @@ where
working_directory_id,
binding.root(),
binding.cwd(),
scope.clone(),
scope,
WorkdirSessionCapabilities::ALL,
binding.command_environment(),
@@ -6817,8 +6801,6 @@ mod tests {
#[test]
fn restored_router_uses_workspace_proxy_for_logical_only_attachment() {
let root = tempfile::tempdir().unwrap();
let scope = manifest::SharedScope::new(manifest::Scope::writable(root.path()).unwrap());
let client = WorkerWorkspaceContext::local_filesystem(Some(
WorkspaceId::new("workspace-a").unwrap(),
))
@@ -6833,7 +6815,6 @@ mod tests {
capabilities: workdir::WorkdirSessionCapabilities::READ_ONLY,
}],
&[],
scope,
client,
)
.unwrap();
@@ -6878,7 +6859,6 @@ mod tests {
.unwrap();
let alias = WorkdirAttachmentAlias::new("docs").unwrap();
let workdir_id = binding.working_directory.id.clone();
let command_output_scope = manifest::SharedScope::new(manifest::Scope::empty());
let attachments = BTreeMap::from([(alias.clone(), binding)]);
let logical = [LogicalWorkdirAttachment {
alias: alias.clone(),
@@ -6890,9 +6870,7 @@ mod tests {
))
.client_handle();
let router =
restored_workdir_router(&attachments, &logical, &[], command_output_scope, client)
.unwrap();
let router = restored_workdir_router(&attachments, &logical, &[], client).unwrap();
let session = router.session(&alias).unwrap();
assert_eq!(
@@ -7579,14 +7557,10 @@ mod tests {
)
.await
.map_err(|err| err.to_string())?;
let bash_output_dir = self.runtime_base.join("bash-output");
let (handle, shutdown_rx) = WorkerController::spawn_runtime_managed(
worker,
&self.runtime_base,
&bash_output_dir,
)
.await
.map_err(|err| err.to_string())?;
let (handle, shutdown_rx) =
WorkerController::spawn_runtime_managed(worker, &self.runtime_base)
.await
.map_err(|err| err.to_string())?;
Ok(RuntimeWorkerController {
handle,
shutdown: Arc::new(tokio::sync::Mutex::new(Some(shutdown_rx))),
@@ -8710,7 +8684,6 @@ mod tests {
"working-directory-42",
root.path(),
root.path(),
spawned_scope.clone(),
spawned_scope,
WorkdirSessionCapabilities::READ_ONLY,
Default::default(),
@@ -8721,7 +8694,6 @@ mod tests {
"working-directory-42",
root.path(),
root.path(),
restored_scope.clone(),
restored_scope,
WorkdirSessionCapabilities::READ_ONLY,
Default::default(),
@@ -10246,9 +10218,7 @@ mod tests {
let alias = WorkdirAttachmentAlias::new("checkout").unwrap();
let attachments = BTreeMap::from([(alias, binding.clone())]);
let scope = manifest::SharedScope::new(manifest::Scope::writable(binding.root()).unwrap());
let router =
runtime_local_workdir_router(&attachments, &BTreeMap::new(), scope, None).unwrap();
let router = runtime_local_workdir_router(&attachments, &BTreeMap::new(), None).unwrap();
session_clone.close().await.unwrap();
assert_eq!(
materializer.cleanup_working_directory(id).unwrap_err().code,