diff --git a/Cargo.lock b/Cargo.lock index df25bfc..c843863 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -864,6 +864,7 @@ dependencies = [ "serde_json", "sha2", "tempfile", + "tokio", "uuid", ] diff --git a/crates/domain/src/approval.rs b/crates/domain/src/approval.rs index 0e62872..29ae7f1 100644 --- a/crates/domain/src/approval.rs +++ b/crates/domain/src/approval.rs @@ -54,7 +54,9 @@ impl ApprovalTargetTool { pub enum ApprovalState { Pending, Granted, - /// Resume has been claimed. Execution may be in flight or interrupted. + /// Resume claimed; waiting for a resource. No side effect is possible yet. + Queued, + /// Resource granted. Mutation dispatch may be in flight or interrupted. Resuming, Denied, Consumed, @@ -65,6 +67,7 @@ impl ApprovalState { match self { Self::Pending => "pending", Self::Granted => "granted", + Self::Queued => "queued", Self::Resuming => "resuming", Self::Denied => "denied", Self::Consumed => "consumed", @@ -75,6 +78,7 @@ impl ApprovalState { match value { "pending" => Ok(Self::Pending), "granted" => Ok(Self::Granted), + "queued" => Ok(Self::Queued), "resuming" => Ok(Self::Resuming), "denied" => Ok(Self::Denied), "consumed" => Ok(Self::Consumed), @@ -143,6 +147,11 @@ mod tests { serde_json::to_value(ApprovalsMode::Confirm).unwrap(), "confirm" ); + assert_eq!(ApprovalState::Queued.as_str(), "queued"); + assert_eq!( + ApprovalState::parse("queued").unwrap(), + ApprovalState::Queued + ); assert_eq!(ApprovalState::Resuming.as_str(), "resuming"); assert_eq!( ApprovalState::parse("resuming").unwrap(), diff --git a/crates/domain/src/error.rs b/crates/domain/src/error.rs index 98e2209..1a6c184 100644 --- a/crates/domain/src/error.rs +++ b/crates/domain/src/error.rs @@ -8,7 +8,10 @@ use serde::{Deserialize, Serialize}; pub enum ErrorCode { Unauthorized, WorkspaceNotFound, + /// Live process owns the workspace mutation lease. WorkspaceBusy, + /// Per-resource waiter queue is at `MAX_WAITERS_PER_RESOURCE`. + ResourceQueueFull, InvalidPatch, InvalidCommand, ProcessSpawnFailed, @@ -39,6 +42,8 @@ pub enum ErrorCode { ApprovalNotFound, ApprovalConflict, ApprovalAmbiguous, + /// Scheduler or store invariant failed. Not occupancy; do not retry as busy. + Internal, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] @@ -134,6 +139,10 @@ mod tests { assert_eq!(json, "\"APPROVAL_CONFLICT\""); let json = serde_json::to_string(&ErrorCode::ApprovalAmbiguous).unwrap(); assert_eq!(json, "\"APPROVAL_AMBIGUOUS\""); + let json = serde_json::to_string(&ErrorCode::ResourceQueueFull).unwrap(); + assert_eq!(json, "\"RESOURCE_QUEUE_FULL\""); + let json = serde_json::to_string(&ErrorCode::Internal).unwrap(); + assert_eq!(json, "\"INTERNAL\""); } #[test] diff --git a/crates/domain/src/execution.rs b/crates/domain/src/execution.rs index 75ca227..75da495 100644 --- a/crates/domain/src/execution.rs +++ b/crates/domain/src/execution.rs @@ -12,6 +12,9 @@ use crate::files::{DEFAULT_FIND_LIMIT, DEFAULT_READ_LIMIT}; pub const PTY_INITIAL_ROWS: u16 = 24; pub const PTY_INITIAL_COLS: u16 = 80; +/// In-memory waiter cap per resource. Excess acquires return `RESOURCE_QUEUE_FULL`. +pub const MAX_WAITERS_PER_RESOURCE: u32 = 64; + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] #[serde(rename_all = "kebab-case")] pub enum ClientEnvironmentKind { @@ -147,14 +150,39 @@ pub struct ProcessExecutionInfo { pub capabilities: Option, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "kebab-case")] +pub enum SerializationScope { + Workspace, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "kebab-case")] +pub enum RequestConflictPolicy { + WaitFifo, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "kebab-case")] +pub enum ProcessConflictPolicy { + Reject, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] pub struct WorkspaceSerializationInfo { + pub scope: SerializationScope, + pub request_conflict: RequestConflictPolicy, + pub process_conflict: ProcessConflictPolicy, + pub queue_durable: bool, + pub max_waiters_per_resource: u32, pub live_process_holds_mutation_lease: bool, pub parallel_exec: bool, pub read_while_process_live: bool, pub find_while_process_live: bool, pub patch_while_process_live: bool, + /// Live process owns the workspace. Not queue saturation. pub conflict_error: ErrorCode, + pub queue_full_error: ErrorCode, } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] @@ -263,12 +291,18 @@ impl WorkspaceExecutionInfo { files, process, serialization: WorkspaceSerializationInfo { + scope: SerializationScope::Workspace, + request_conflict: RequestConflictPolicy::WaitFifo, + process_conflict: ProcessConflictPolicy::Reject, + queue_durable: false, + max_waiters_per_resource: MAX_WAITERS_PER_RESOURCE, live_process_holds_mutation_lease: true, parallel_exec: false, read_while_process_live: true, find_while_process_live: true, patch_while_process_live: false, conflict_error: ErrorCode::WorkspaceBusy, + queue_full_error: ErrorCode::ResourceQueueFull, }, isolation: IsolationInfo { file_tools_workspace_scoped: true, @@ -399,9 +433,36 @@ mod tests { assert_eq!(exec.network.policy, NetworkPolicyState::Restricted); assert_eq!(exec.network.enforcement, NetworkEnforcementState::None); assert!(!exec.network.client_may_escalate); + assert_eq!(exec.serialization.scope, SerializationScope::Workspace); + assert_eq!( + exec.serialization.request_conflict, + RequestConflictPolicy::WaitFifo + ); + assert_eq!( + exec.serialization.process_conflict, + ProcessConflictPolicy::Reject + ); + assert!(!exec.serialization.queue_durable); + assert_eq!( + exec.serialization.max_waiters_per_resource, + MAX_WAITERS_PER_RESOURCE + ); assert_eq!(exec.serialization.conflict_error, ErrorCode::WorkspaceBusy); + assert_eq!( + exec.serialization.queue_full_error, + ErrorCode::ResourceQueueFull + ); let json = serde_json::to_value(&exec).unwrap(); + assert_eq!(json["serialization"]["scope"], "workspace"); + assert_eq!(json["serialization"]["request_conflict"], "wait-fifo"); + assert_eq!(json["serialization"]["process_conflict"], "reject"); + assert_eq!(json["serialization"]["queue_durable"], false); + assert_eq!(json["serialization"]["max_waiters_per_resource"], 64); assert_eq!(json["serialization"]["conflict_error"], "WORKSPACE_BUSY"); + assert_eq!( + json["serialization"]["queue_full_error"], + "RESOURCE_QUEUE_FULL" + ); assert!(json.get("environment_id").is_none()); assert_eq!(json["environment"]["kind"], "host"); assert_eq!(json["environment"]["file_read_supported"], true); diff --git a/crates/domain/src/lib.rs b/crates/domain/src/lib.rs index 6d7b6e4..1f86d8f 100644 --- a/crates/domain/src/lib.rs +++ b/crates/domain/src/lib.rs @@ -25,9 +25,10 @@ pub use execution::{ ClientEnvironmentKind, CommandSandboxState, EffectivePermissionInfo, EnvironmentExecutionInfo, FileCapabilityInfo, FileExecutionInfo, FileOperationInfo, IsolationInfo, NetworkEnforcementState, NetworkInfo, NetworkPolicyState, ProcessCapabilityInfo, - ProcessDisconnectAction, ProcessExecutionInfo, ProcessLifetimeInfo, ProcessLifetimeOwner, - ProcessRestartRecovery, PtyCapabilityInfo, WorkspaceExecutionInfo, WorkspaceSerializationInfo, - PTY_INITIAL_COLS, PTY_INITIAL_ROWS, + ProcessConflictPolicy, ProcessDisconnectAction, ProcessExecutionInfo, ProcessLifetimeInfo, + ProcessLifetimeOwner, ProcessRestartRecovery, PtyCapabilityInfo, RequestConflictPolicy, + SerializationScope, WorkspaceExecutionInfo, WorkspaceSerializationInfo, + MAX_WAITERS_PER_RESOURCE, PTY_INITIAL_COLS, PTY_INITIAL_ROWS, }; pub use files::{ FindParams, FindResult, ReadParams, ReadResult, DEFAULT_FIND_LIMIT, DEFAULT_READ_LIMIT, diff --git a/crates/server/src/mcp.rs b/crates/server/src/mcp.rs index fcf0e77..794f6a8 100644 --- a/crates/server/src/mcp.rs +++ b/crates/server/src/mcp.rs @@ -74,8 +74,11 @@ workspace cwd and returns a server-minted process_id. Ending an MCP request \ does not terminate the process. A live managed process holds the workspace mutation lease. read and find may \ -continue, but apply_patch or another exec_command may return WORKSPACE_BUSY \ -until the process exits or is terminated. +continue, but apply_patch or another exec_command returns WORKSPACE_BUSY \ +until the process exits or is terminated. Request-owned patch or exec work \ +on the same workspace waits in FIFO order until acquire(); a live process \ +fails already-queued waiters with WORKSPACE_BUSY. Queue saturation returns \ +RESOURCE_QUEUE_FULL. exec_command.tty is optional and defaults to false. tty=true attaches a \ fixed 24x80 PTY. Use process_resize on a running PTY to change rows and \ @@ -170,7 +173,7 @@ impl CodeSpace { #[tool( name = "workspace_info", - description = "Return CodeSpace identity and, when workspace_id is set, the effective execution contract. files.*.available and process.available reflect permission and backend support only. They do not include transient workspace occupancy; exec_command or apply_patch may still return WORKSPACE_BUSY. Tool existence is reported separately by tools_exposed. Does not call a model. Does not read files. workspace_id is a selector, not a credential." + description = "Return CodeSpace identity and, when workspace_id is set, the effective execution contract. files.*.available and process.available reflect permission and backend support only. They do not include transient workspace occupancy; exec_command or apply_patch may still return WORKSPACE_BUSY or RESOURCE_QUEUE_FULL. serialization reports request_conflict wait-fifo, process_conflict reject, and max_waiters_per_resource. Tool existence is reported separately by tools_exposed. Does not call a model. Does not read files. workspace_id is a selector, not a credential." )] async fn workspace_info( &self, @@ -227,7 +230,7 @@ impl CodeSpace { #[tool( name = "apply_patch", - description = "Apply a Codex V4A patch. check_only verifies without writing and returns status checked. status applied means disk hashes match the helper claim. Never falls back to git apply. status=unknown means the mutation may have executed but its result could not be confirmed. Do not retry the same mutation under a new operation_key. operation_key provides replay/idempotency for the same logical mutation. When the workspace approvals mode is confirm, a policy-allowed request returns APPROVAL_REQUIRED before begin() and does not write." + description = "Apply a Codex V4A patch. check_only verifies without writing and returns status checked. status applied means disk hashes match the helper claim. Never falls back to git apply. status=unknown means the mutation may have executed but its result could not be confirmed. Do not retry the same mutation under a new operation_key. operation_key provides replay/idempotency for the same logical mutation. When the workspace approvals mode is confirm, a policy-allowed request returns APPROVAL_REQUIRED before begin() and does not write. A live process returns WORKSPACE_BUSY and fails waiters that were already queued. Overlapping request-owned patch or exec work on the same workspace waits in FIFO order starting at acquire(). Queue saturation returns RESOURCE_QUEUE_FULL." )] async fn apply_patch( &self, @@ -242,7 +245,7 @@ impl CodeSpace { if let Err(err) = self.maybe_hold_patch(ws, ¶ms) { return Err(err_json(err)); } - self.apply_patch_inner(params) + self.apply_patch_inner(params, None) .await .map(Json) .map_err(err_json) @@ -264,7 +267,7 @@ impl CodeSpace { #[tool( name = "exec_command", - description = "Start a managed argv in the workspace cwd. There is no implicit shell. Returns a server-minted process_id and a dispatch_status. Request end does not terminate the process. Omitted or false tty uses pipes. tty=true attaches a 24x80 PTY; use process_resize to change the size of a running PTY. tty_size is not an exec_command argument. Use tty only for commands requiring terminal semantics or an interactive TUI. A live process holds the workspace mutation lease, so another exec_command or apply_patch may return WORKSPACE_BUSY until it exits or is terminated. Use write_stdin, read_process, process_status, process_resize, and terminate_process with the returned process_id. dispatch_status=unknown means the spawn may have occurred. Do not blindly start a duplicate process. The returned process_id identifies the uncertain attempt. Use read_process, process_status, process_resize, or terminate_process when the backend remains reachable; do not assume that unknown means the process did not start. PROCESS_SPAWN_FAILED means the backend confirmed that no managed process was started; it is distinct from dispatch_status=unknown. When the workspace approvals mode is confirm, a policy-allowed request returns APPROVAL_REQUIRED before spawn." + description = "Start a managed argv in the workspace cwd. There is no implicit shell. Returns a server-minted process_id and a dispatch_status. Request end does not terminate the process. Omitted or false tty uses pipes. tty=true attaches a 24x80 PTY; use process_resize to change the size of a running PTY. tty_size is not an exec_command argument. Use tty only for commands requiring terminal semantics or an interactive TUI. A live process holds the workspace mutation lease, so another exec_command or apply_patch returns WORKSPACE_BUSY until it exits or is terminated, including waiters that were already queued. Overlapping request-owned patch and exec work on the same workspace waits in FIFO order starting at acquire(). Queue saturation returns RESOURCE_QUEUE_FULL. Use write_stdin, read_process, process_status, process_resize, and terminate_process with the returned process_id. dispatch_status=unknown means the spawn may have occurred. Do not blindly start a duplicate process. The returned process_id identifies the uncertain attempt. Use read_process, process_status, process_resize, or terminate_process when the backend remains reachable; do not assume that unknown means the process did not start. PROCESS_SPAWN_FAILED means the backend confirmed that no managed process was started; it is distinct from dispatch_status=unknown. When the workspace approvals mode is confirm, a policy-allowed request returns APPROVAL_REQUIRED before spawn." )] async fn exec_command( &self, @@ -282,7 +285,7 @@ impl CodeSpace { if let Err(err) = self.maybe_hold_exec(ws, ¶ms) { return Err(err_json(err)); } - self.exec_command_inner(params) + self.exec_command_inner(params, None) .await .map(Json) .map_err(err_json) @@ -519,11 +522,15 @@ impl CodeSpace { async fn apply_patch_inner( &self, params: ApplyPatchParams, + resume: Option<&ApprovalId>, ) -> Result { let ws = self.registry.get(¶ms.workspace_id.0)?; codespace_policy::allow(ws, Action::Write, &ClientClaims::default())?; ws.require_file_write()?; - let _lease = self.store.try_acquire_write(¶ms.workspace_id.0)?; + let _lease = self.store.acquire_write(¶ms.workspace_id.0).await?; + if let Some(id) = resume { + self.store.mark_resuming(id)?; + } let fingerprint = Store::fingerprint(¶ms); match self.store.begin( params.operation_key.as_ref(), @@ -804,7 +811,9 @@ impl CodeSpace { ApprovalTargetTool::ApplyPatch => { let params: ApplyPatchParams = serde_json::from_str(&record.params_json) .map_err(|err| ErrorBody::new(ErrorCode::InvalidPatch, err.to_string()))?; - let result = self.apply_patch_inner(params).await?; + let result = self + .apply_patch_inner(params, Some(&record.approval_id)) + .await?; Ok(OperationResumeResult { approval_id: record.approval_id, state: ApprovalState::Consumed, @@ -815,7 +824,9 @@ impl CodeSpace { ApprovalTargetTool::ExecCommand => { let params: ExecCommandParams = serde_json::from_str(&record.params_json) .map_err(|err| ErrorBody::new(ErrorCode::InvalidCommand, err.to_string()))?; - let result = self.exec_command_inner(params).await?; + let result = self + .exec_command_inner(params, Some(&record.approval_id)) + .await?; Ok(OperationResumeResult { approval_id: record.approval_id, state: ApprovalState::Consumed, @@ -829,6 +840,7 @@ impl CodeSpace { async fn exec_command_inner( &self, params: ExecCommandParams, + resume: Option<&ApprovalId>, ) -> Result { let ws = self.registry.get(¶ms.workspace_id.0)?; allow(ws, Action::Exec, &ClientClaims::default())?; @@ -837,24 +849,36 @@ impl CodeSpace { return Err(invalid_argv()); } let process_id = ProcessId(format!("proc-{}", Uuid::new_v4())); - self.store - .mark_shell_busy(¶ms.workspace_id.0, &process_id.0)?; + let mut reservation = self + .store + .acquire_shell_busy(¶ms.workspace_id.0, &process_id.0) + .await?; + if let Some(id) = resume { + self.store.mark_resuming(id)?; + } + reservation.arm_dispatch(); let mut req = RunnerExecRequest::for_host(params.command, process_id.clone(), ws.profile); req.policy.network = ws.network; req.tty = params.tty; match self.runner.exec(ws, req).await { - Ok(result) => Ok(ExecCommandResult { - process_id: result.process_id, - dispatch_status: ExecDispatchStatus::Confirmed, - coordination: self.hint(¶ms.workspace_id.0, params.work_id.as_ref()), - }), - Err(RunnerError::TransportAmbiguous { .. }) => Ok(ExecCommandResult { - process_id, - dispatch_status: ExecDispatchStatus::Unknown, - coordination: self.hint(¶ms.workspace_id.0, params.work_id.as_ref()), - }), + Ok(result) => { + reservation.confirm(); + Ok(ExecCommandResult { + process_id: result.process_id, + dispatch_status: ExecDispatchStatus::Confirmed, + coordination: self.hint(¶ms.workspace_id.0, params.work_id.as_ref()), + }) + } + Err(RunnerError::TransportAmbiguous { .. }) => { + reservation.confirm(); + Ok(ExecCommandResult { + process_id, + dispatch_status: ExecDispatchStatus::Unknown, + coordination: self.hint(¶ms.workspace_id.0, params.work_id.as_ref()), + }) + } Err(err) => { - self.store.release_process(&process_id.0); + reservation.abort(); Err(err.into_error_body()) } } @@ -1087,6 +1111,7 @@ mod tests { assert!(text.contains("PROCESS_NOT_TTY"), "{text}"); assert!(text.contains("PROCESS_NOT_RUNNING"), "{text}"); assert!(text.contains("WORKSPACE_BUSY"), "{text}"); + assert!(text.contains("RESOURCE_QUEUE_FULL"), "{text}"); assert!(text.contains("command_sandbox is linux-sandbox"), "{text}"); assert!( text.contains("host execution is not an OS command sandbox"), @@ -1215,6 +1240,28 @@ mod tests { json["execution"]["files"]["capabilities"]["find_max_paths"], 10000 ); + assert_eq!(json["execution"]["serialization"]["scope"], "workspace"); + assert_eq!( + json["execution"]["serialization"]["request_conflict"], + "wait-fifo" + ); + assert_eq!( + json["execution"]["serialization"]["process_conflict"], + "reject" + ); + assert_eq!(json["execution"]["serialization"]["queue_durable"], false); + assert_eq!( + json["execution"]["serialization"]["max_waiters_per_resource"], + 64 + ); + assert_eq!( + json["execution"]["serialization"]["conflict_error"], + "WORKSPACE_BUSY" + ); + assert_eq!( + json["execution"]["serialization"]["queue_full_error"], + "RESOURCE_QUEUE_FULL" + ); assert!(json.get("environment_id").is_none()); assert!(!json.to_string().contains("\"environment_id\"")); } @@ -1361,14 +1408,17 @@ mod tests { let runner = RuntimeBackend::Uds(UdsRunner::from_stream(client, Arc::new(|_| {}))); let cs = CodeSpace::with_store_and_runner(write_registry(ws_root), store.clone(), runner); let result = cs - .apply_patch_inner(ApplyPatchParams { - workspace_id: WorkspaceId("demo".into()), - patch: "*** Begin Patch\n*** Add File: lost.txt\n+x\n*** End Patch\n".into(), - expected_versions: BTreeMap::new(), - operation_key: Some(OperationKey("k-unknown".into())), - check_only: false, - work_id: None, - }) + .apply_patch_inner( + ApplyPatchParams { + workspace_id: WorkspaceId("demo".into()), + patch: "*** Begin Patch\n*** Add File: lost.txt\n+x\n*** End Patch\n".into(), + expected_versions: BTreeMap::new(), + operation_key: Some(OperationKey("k-unknown".into())), + check_only: false, + work_id: None, + }, + None, + ) .await .unwrap(); assert_eq!(result.status, PatchStatus::Unknown); @@ -1408,14 +1458,17 @@ mod tests { assert!(started.0.process_id.0.starts_with("proc-")); assert_eq!(started.0.dispatch_status, ExecDispatchStatus::Unknown); let busy = cs - .apply_patch_inner(ApplyPatchParams { - workspace_id: WorkspaceId("demo".into()), - patch: "*** Begin Patch\n*** Add File: later.txt\n+x\n*** End Patch\n".into(), - expected_versions: BTreeMap::new(), - operation_key: None, - check_only: false, - work_id: None, - }) + .apply_patch_inner( + ApplyPatchParams { + workspace_id: WorkspaceId("demo".into()), + patch: "*** Begin Patch\n*** Add File: later.txt\n+x\n*** End Patch\n".into(), + expected_versions: BTreeMap::new(), + operation_key: None, + check_only: false, + work_id: None, + }, + None, + ) .await .unwrap_err(); assert_eq!(busy.code, ErrorCode::WorkspaceBusy); @@ -1478,6 +1531,44 @@ mod tests { assert!(started.0.process_id.0.starts_with("proc-")); } + #[tokio::test] + async fn aborted_apply_waiter_releases_queue_slot() { + let dir = tempfile::tempdir().unwrap(); + let ws_root = dir.path().join("ws"); + std::fs::create_dir(&ws_root).unwrap(); + let store = Arc::new(Store::memory().unwrap()); + let cs = CodeSpace::with_store(write_registry(ws_root), store.clone()); + let lease = store.acquire_write("demo").await.unwrap(); + let cs_wait = cs.clone(); + let waiting = tokio::spawn(async move { + cs_wait + .apply_patch_inner( + ApplyPatchParams { + workspace_id: WorkspaceId("demo".into()), + patch: "*** Begin Patch\n*** Add File: wait.txt\n+x\n*** End Patch\n" + .into(), + expected_versions: BTreeMap::new(), + operation_key: Some(OperationKey("wait-cancel".into())), + check_only: false, + work_id: None, + }, + None, + ) + .await + }); + for _ in 0..30 { + tokio::task::yield_now().await; + std::thread::sleep(std::time::Duration::from_millis(5)); + } + waiting.abort(); + let _ = waiting.await; + drop(lease); + let _next = store + .acquire_write("demo") + .await + .expect("queue released after cancelled waiter"); + } + fn parse_exec_err(result: Result, String>) -> ErrorBody { match result { Err(err) => serde_json::from_str(&err) @@ -1604,14 +1695,17 @@ mod tests { registry.insert(ws); let cs = CodeSpace::new(registry); let err = cs - .apply_patch_inner(ApplyPatchParams { - workspace_id: WorkspaceId("demo".into()), - patch: "*** Begin Patch\n*** Add File: a.txt\n+x\n*** End Patch\n".into(), - expected_versions: BTreeMap::new(), - operation_key: Some(OperationKey("k-box".into())), - check_only: false, - work_id: None, - }) + .apply_patch_inner( + ApplyPatchParams { + workspace_id: WorkspaceId("demo".into()), + patch: "*** Begin Patch\n*** Add File: a.txt\n+x\n*** End Patch\n".into(), + expected_versions: BTreeMap::new(), + operation_key: Some(OperationKey("k-box".into())), + check_only: false, + work_id: None, + }, + None, + ) .await .unwrap_err(); assert_eq!(err.code, ErrorCode::Unauthorized); diff --git a/crates/server/tests/apply.rs b/crates/server/tests/apply.rs index 73779e6..e45fc2c 100644 --- a/crates/server/tests/apply.rs +++ b/crates/server/tests/apply.rs @@ -13,6 +13,18 @@ fn payload(result: &rmcp::model::CallToolResult) -> serde_json::Value { }) } +fn err_text(result: &Result) -> String { + match result { + Ok(r) => r + .content + .iter() + .filter_map(|c| c.as_text().map(|t| t.text.clone())) + .collect::>() + .join(""), + Err(err) => err.to_string(), + } +} + #[tokio::test] async fn apply_writes_and_check_only_does_not() { let root = tempfile::tempdir().unwrap(); @@ -115,3 +127,207 @@ async fn apply_writes_and_check_only_does_not() { client.cancel().await.expect("cancel"); } + +#[tokio::test] +async fn concurrent_apply_patch_serializes_without_busy() { + let root = tempfile::tempdir().unwrap(); + let ws = root.path().join("ws"); + std::fs::create_dir(&ws).unwrap(); + let cfg = root.path().join("workspaces.json"); + std::fs::write( + &cfg, + serde_json::json!({ + "workspaces": { + "demo": { "root": ws, "profile": "workspace-write" } + } + }) + .to_string(), + ) + .unwrap(); + let db = root.path().join("ops.sqlite"); + let bin = env!("CARGO_BIN_EXE_codespace-mcp"); + let patch_bin = codespace_server::patch_helper::ensure_helper_for_tests(); + let client = () + .serve( + TokioChildProcess::new(Command::new(bin).configure(|cmd| { + cmd.env("CODESPACE_CONFIG", &cfg) + .env("CODESPACE_OPERATIONS_DB", &db) + .env("CODESPACE_PATCH_BIN", &patch_bin); + })) + .expect("spawn"), + ) + .await + .expect("init"); + + let first = client.call_tool(CallToolRequestParams::new(TOOL_APPLY_PATCH).with_arguments( + object!({ + "workspace_id": "demo", + "patch": "*** Begin Patch\n*** Add File: a.txt\n+one\n*** End Patch\n", + "operation_key": "sched-a" + }), + )); + let second = client.call_tool(CallToolRequestParams::new(TOOL_APPLY_PATCH).with_arguments( + object!({ + "workspace_id": "demo", + "patch": "*** Begin Patch\n*** Add File: b.txt\n+two\n*** End Patch\n", + "operation_key": "sched-b" + }), + )); + let (first, second) = tokio::join!(first, second); + let first_body = payload(&first.expect("first patch")); + let second_body = payload(&second.expect("second patch")); + assert_eq!(first_body["status"], "applied"); + assert_eq!(second_body["status"], "applied"); + assert_eq!( + std::fs::read_to_string(root.path().join("ws/a.txt")).unwrap(), + "one\n" + ); + assert_eq!( + std::fs::read_to_string(root.path().join("ws/b.txt")).unwrap(), + "two\n" + ); + + client.cancel().await.expect("cancel"); +} + +#[tokio::test] +async fn concurrent_same_expected_version_is_version_conflict() { + let root = tempfile::tempdir().unwrap(); + let ws = root.path().join("ws"); + std::fs::create_dir(&ws).unwrap(); + std::fs::write(ws.join("target.txt"), "v1\n").unwrap(); + let cfg = root.path().join("workspaces.json"); + std::fs::write( + &cfg, + serde_json::json!({ + "workspaces": { + "demo": { "root": ws, "profile": "workspace-write" } + } + }) + .to_string(), + ) + .unwrap(); + let db = root.path().join("ops.sqlite"); + let bin = env!("CARGO_BIN_EXE_codespace-mcp"); + let patch_bin = codespace_server::patch_helper::ensure_helper_for_tests(); + let client = () + .serve( + TokioChildProcess::new(Command::new(bin).configure(|cmd| { + cmd.env("CODESPACE_CONFIG", &cfg) + .env("CODESPACE_OPERATIONS_DB", &db) + .env("CODESPACE_PATCH_BIN", &patch_bin); + })) + .expect("spawn"), + ) + .await + .expect("init"); + + let read = client + .call_tool( + CallToolRequestParams::new(TOOL_READ) + .with_arguments(object!({ "workspace_id": "demo", "path": "target.txt" })), + ) + .await + .expect("read"); + let version = payload(&read)["version"].as_str().unwrap().to_string(); + let first = client.call_tool(CallToolRequestParams::new(TOOL_APPLY_PATCH).with_arguments( + object!({ + "workspace_id": "demo", + "expected_versions": { "target.txt": version }, + "patch": "*** Begin Patch\n*** Update File: target.txt\n@@\n-v1\n+a\n*** End Patch\n", + "operation_key": "ver-a" + }), + )); + let second = client.call_tool(CallToolRequestParams::new(TOOL_APPLY_PATCH).with_arguments( + object!({ + "workspace_id": "demo", + "expected_versions": { "target.txt": version }, + "patch": "*** Begin Patch\n*** Update File: target.txt\n@@\n-v1\n+b\n*** End Patch\n", + "operation_key": "ver-b" + }), + )); + let (first, second) = tokio::join!(first, second); + let texts = [err_text(&first), err_text(&second)]; + let applied = [ + first.as_ref().ok().map(payload), + second.as_ref().ok().map(payload), + ] + .into_iter() + .flatten() + .filter(|body| body["status"] == "applied") + .count(); + let conflicts = texts + .iter() + .filter(|text| text.contains("VERSION_CONFLICT")) + .count(); + assert_eq!(applied, 1, "first={:?} second={:?}", texts[0], texts[1]); + assert_eq!(conflicts, 1, "first={:?} second={:?}", texts[0], texts[1]); + let on_disk = std::fs::read_to_string(root.path().join("ws/target.txt")).unwrap(); + assert!(on_disk == "a\n" || on_disk == "b\n", "{on_disk}"); + + client.cancel().await.expect("cancel"); +} + +#[tokio::test] +async fn concurrent_same_operation_key_replays() { + let root = tempfile::tempdir().unwrap(); + let ws = root.path().join("ws"); + std::fs::create_dir(&ws).unwrap(); + let cfg = root.path().join("workspaces.json"); + std::fs::write( + &cfg, + serde_json::json!({ + "workspaces": { + "demo": { "root": ws, "profile": "workspace-write" } + } + }) + .to_string(), + ) + .unwrap(); + let db = root.path().join("ops.sqlite"); + let bin = env!("CARGO_BIN_EXE_codespace-mcp"); + let patch_bin = codespace_server::patch_helper::ensure_helper_for_tests(); + let client = () + .serve( + TokioChildProcess::new(Command::new(bin).configure(|cmd| { + cmd.env("CODESPACE_CONFIG", &cfg) + .env("CODESPACE_OPERATIONS_DB", &db) + .env("CODESPACE_PATCH_BIN", &patch_bin); + })) + .expect("spawn"), + ) + .await + .expect("init"); + + let patch = "*** Begin Patch\n*** Add File: once.txt\n+hello\n*** End Patch\n"; + let first = client.call_tool(CallToolRequestParams::new(TOOL_APPLY_PATCH).with_arguments( + object!({ + "workspace_id": "demo", + "patch": patch, + "operation_key": "same-key" + }), + )); + let second = client.call_tool(CallToolRequestParams::new(TOOL_APPLY_PATCH).with_arguments( + object!({ + "workspace_id": "demo", + "patch": patch, + "operation_key": "same-key" + }), + )); + let (first, second) = tokio::join!(first, second); + let first_body = payload(&first.expect("first")); + let second_body = payload(&second.expect("second")); + assert_eq!(first_body["status"], "applied"); + assert_eq!(second_body["status"], "applied"); + assert_eq!(first_body["operation_id"], second_body["operation_id"]); + assert!( + first_body["replayed"] == true || second_body["replayed"] == true, + "first={first_body} second={second_body}" + ); + assert_eq!( + std::fs::read_to_string(root.path().join("ws/once.txt")).unwrap(), + "hello\n" + ); + + client.cancel().await.expect("cancel"); +} diff --git a/crates/server/tests/approvals.rs b/crates/server/tests/approvals.rs index f9f15c6..939e17d 100644 --- a/crates/server/tests/approvals.rs +++ b/crates/server/tests/approvals.rs @@ -723,6 +723,50 @@ async fn restart_resuming_exec_is_ambiguous_without_spawn() { client.cancel().await.expect("cancel"); } +#[tokio::test] +async fn restart_queued_exec_reacquires_and_runs() { + let (root, cfg, ws) = write_workspace("workspace-write", Some("confirm")); + let db = root.path().join("ops.sqlite"); + let client = spawn_client(&cfg, &db).await; + let held = client + .call_tool( + CallToolRequestParams::new(TOOL_EXEC_COMMAND).with_arguments(object!({ + "workspace_id": "demo", + "command": ["/bin/sh", "-c", "printf x > held.txt"] + })), + ) + .await; + let approval_id = error_body(&held)["approval_id"] + .as_str() + .unwrap() + .to_string(); + grant(&client, &approval_id).await; + client.cancel().await.expect("cancel"); + force_approval_state(&db, &approval_id, ApprovalState::Queued, true); + + let client = spawn_client(&cfg, &db).await; + let resumed = client + .call_tool( + CallToolRequestParams::new(TOOL_OPERATION_RESUME) + .with_arguments(object!({ "approval_id": approval_id })), + ) + .await + .expect("queued exec resume"); + assert_eq!( + payload(&resumed)["exec_command"]["dispatch_status"], + "confirmed" + ); + let held_path = ws.join("held.txt"); + for _ in 0..50 { + if held_path.exists() { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + assert_eq!(std::fs::read_to_string(held_path).unwrap(), "x"); + client.cancel().await.expect("cancel"); +} + #[tokio::test] async fn restart_preserves_pending_granted_and_consumed() { let (root, cfg, ws) = write_workspace("workspace-write", Some("confirm")); diff --git a/crates/server/tests/process.rs b/crates/server/tests/process.rs index a9c3c2e..20f3de2 100644 --- a/crates/server/tests/process.rs +++ b/crates/server/tests/process.rs @@ -778,6 +778,7 @@ async fn exec_command_schema_has_optional_tty_and_live_tools_unchanged() { "{exec_desc}" ); assert!(exec_desc.contains("WORKSPACE_BUSY"), "{exec_desc}"); + assert!(exec_desc.contains("RESOURCE_QUEUE_FULL"), "{exec_desc}"); assert!( exec_desc.contains("uncertain attempt") || exec_desc.contains("backend remains reachable"), "{exec_desc}" @@ -871,6 +872,16 @@ async fn exec_command_schema_has_optional_tty_and_live_tools_unchanged() { exec["process"]["capabilities"]["lifetime"]["restart_recovery"], "none" ); + assert_eq!(exec["serialization"]["scope"], "workspace"); + assert_eq!(exec["serialization"]["request_conflict"], "wait-fifo"); + assert_eq!(exec["serialization"]["process_conflict"], "reject"); + assert_eq!(exec["serialization"]["queue_durable"], false); + assert_eq!(exec["serialization"]["max_waiters_per_resource"], 64); + assert_eq!(exec["serialization"]["conflict_error"], "WORKSPACE_BUSY"); + assert_eq!( + exec["serialization"]["queue_full_error"], + "RESOURCE_QUEUE_FULL" + ); if codespace_runner::linux_sandbox_available() { assert_eq!(exec["isolation"]["command_sandbox"], "linux-sandbox"); assert_eq!(exec["network"]["enforcement"], "enforced"); diff --git a/crates/store/Cargo.toml b/crates/store/Cargo.toml index b5ad65f..f645487 100644 --- a/crates/store/Cargo.toml +++ b/crates/store/Cargo.toml @@ -5,7 +5,7 @@ edition.workspace = true license.workspace = true rust-version.workspace = true publish = false -description = "SQLite operations plus in-memory resource serialization" +description = "SQLite operations plus in-memory resource occupancy and FIFO scheduling" [dependencies] codespace-domain = { path = "../domain" } @@ -15,6 +15,8 @@ serde_json = { workspace = true } sha2 = "0.10" hex = "0.4" uuid = { version = "1", features = ["v4"] } +tokio = { version = "1", features = ["sync"] } [dev-dependencies] tempfile = "3" +tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync"] } diff --git a/crates/store/src/approvals.rs b/crates/store/src/approvals.rs index da6253a..7809e76 100644 --- a/crates/store/src/approvals.rs +++ b/crates/store/src/approvals.rs @@ -193,7 +193,7 @@ impl Store { "consumed" => return replay_consumed(approval_id, row.result_json.as_deref()), "pending" => return Err(conflict("approval is still pending")), "denied" => return Err(conflict("approval was denied")), - "granted" | "resuming" => { + "granted" | "queued" | "resuming" => { drop(conn); let guard = self.mark_inflight(&approval_id.0)?; let conn = self.conn.lock().expect("sqlite mutex"); @@ -210,7 +210,7 @@ impl Store { let updated = conn .execute( "UPDATE approvals - SET state = 'resuming' + SET state = 'queued' WHERE approval_id = ?1 AND state = 'granted'", params![approval_id.0], ) @@ -222,6 +222,9 @@ impl Store { } return Ok(ResumeClaim::Execute(row.into_record()?, guard)); } + "queued" => { + return Ok(ResumeClaim::Execute(row.into_record()?, guard)); + } "resuming" => { return Ok(ResumeClaim::Reconcile(row.into_record()?, guard)); } @@ -253,6 +256,25 @@ impl Store { approval_id: &ApprovalId, result: Result<&OperationResumeResult, &ErrorBody>, ) -> Result<(), ErrorBody> { + let queued = { + let conn = self.conn.lock().expect("sqlite mutex"); + load(&conn, &approval_id.0)? + .ok_or_else(|| not_found(approval_id))? + .state + == ApprovalState::Queued.as_str() + }; + if queued + && matches!( + &result, + Err(err) + if matches!( + err.code, + ErrorCode::WorkspaceBusy | ErrorCode::ResourceQueueFull + ) + ) + { + return Ok(()); + } if self.take_fail_next_finish() { return Err(ambiguous( approval_id, @@ -276,7 +298,7 @@ impl Store { .execute( "UPDATE approvals SET state = 'consumed', result_json = ?1, params_json = ?2 - WHERE approval_id = ?3 AND state = 'resuming'", + WHERE approval_id = ?3 AND state IN ('queued', 'resuming')", params![json, scrubbed, approval_id.0], ) .map_err(db_err)?; @@ -290,6 +312,31 @@ impl Store { Ok(()) } + pub fn mark_resuming(&self, approval_id: &ApprovalId) -> Result<(), ErrorBody> { + let conn = self.conn.lock().expect("sqlite mutex"); + let row = load(&conn, &approval_id.0)?.ok_or_else(|| not_found(approval_id))?; + match row.state.as_str() { + "resuming" => Ok(()), + "queued" => { + let updated = conn + .execute( + "UPDATE approvals + SET state = 'resuming' + WHERE approval_id = ?1 AND state = 'queued'", + params![approval_id.0], + ) + .map_err(db_err)?; + if updated != 1 { + return Err(conflict("approval is not queued for dispatch")); + } + Ok(()) + } + other => Err(conflict(format!( + "approval is in an unknown state `{other}`" + ))), + } + } + pub fn fail_next_finish(&self) { *self.fail_next_finish.lock().expect("fail flag") = true; } @@ -372,7 +419,7 @@ fn load_active_by_fingerprint( "SELECT approval_id, workspace_id, tool, fingerprint, params_json, state, resolved_at, result_json FROM approvals WHERE workspace_id = ?1 AND fingerprint = ?2 - AND state IN ('pending','granted','resuming') + AND state IN ('pending','granted','queued','resuming') LIMIT 1", params![workspace_id, fingerprint], map_row, @@ -636,6 +683,7 @@ mod tests { ResumeClaim::Execute(_, guard) => guard, other => panic!("expected execute, got {other:?}"), }; + store.mark_resuming(&created.approval_id).unwrap(); store.fail_next_finish(); assert_eq!( store @@ -668,6 +716,7 @@ mod tests { ResumeClaim::Execute(_, guard) => guard, other => panic!("expected execute, got {other:?}"), }); + store.mark_resuming(&created.approval_id).unwrap(); match store.claim_resume(&created.approval_id).unwrap() { ResumeClaim::Reconcile(record, _guard) => { assert_eq!(record.state, ApprovalState::Resuming); @@ -676,6 +725,79 @@ mod tests { }; } + #[test] + fn restart_queued_is_execute() { + let store = Store::memory().unwrap(); + let params = patch_params(); + let created = store + .create_approval( + "demo", + ApprovalTargetTool::ApplyPatch, + &Store::fingerprint(¶ms), + serde_json::to_value(¶ms).unwrap(), + ) + .unwrap(); + store + .resolve_approval(&created.approval_id, ApprovalDecision::Grant) + .unwrap(); + drop(match store.claim_resume(&created.approval_id).unwrap() { + ResumeClaim::Execute(_, guard) => guard, + other => panic!("expected execute, got {other:?}"), + }); + let (state, _, _) = store.inspect_approval(&created.approval_id).unwrap(); + assert_eq!(state, ApprovalState::Queued); + match store.claim_resume(&created.approval_id).unwrap() { + ResumeClaim::Execute(record, _guard) => { + assert_eq!(record.tool, ApprovalTargetTool::ApplyPatch); + } + other => panic!("expected execute, got {other:?}"), + }; + } + + #[test] + fn queued_scheduler_errors_do_not_consume_approval() { + for (index, code) in [ErrorCode::WorkspaceBusy, ErrorCode::ResourceQueueFull] + .into_iter() + .enumerate() + { + let store = Store::memory().unwrap(); + let created = store + .create_approval( + "demo", + ApprovalTargetTool::ExecCommand, + &format!("sha256:retryable-{index}"), + json!({"workspace_id":"demo","command":["/bin/echo","ok"]}), + ) + .unwrap(); + store + .resolve_approval(&created.approval_id, ApprovalDecision::Grant) + .unwrap(); + + let guard = match store.claim_resume(&created.approval_id).unwrap() { + ResumeClaim::Execute(_, guard) => guard, + other => panic!("expected execute, got {other:?}"), + }; + let err = ErrorBody::new(code, "transient scheduler conflict"); + store + .finish_resume(&created.approval_id, Err(&err)) + .unwrap(); + + let (state, params_json, result_json) = + store.inspect_approval(&created.approval_id).unwrap(); + assert_eq!(state, ApprovalState::Queued); + assert!(!params_json.contains("\"scrubbed\":true")); + assert!(result_json.is_none()); + + drop(guard); + match store.claim_resume(&created.approval_id).unwrap() { + ResumeClaim::Execute(record, _guard) => { + assert_eq!(record.state, ApprovalState::Queued); + } + other => panic!("expected retryable execute, got {other:?}"), + }; + } + } + #[test] fn consumed_fingerprint_can_open_a_new_hold() { let store = Store::memory().unwrap(); diff --git a/crates/store/src/lib.rs b/crates/store/src/lib.rs index 96f8194..6478725 100644 --- a/crates/store/src/lib.rs +++ b/crates/store/src/lib.rs @@ -1,4 +1,8 @@ -//! Single-instance SQLite operations, confirmation holds, and in-process workspace write locks. +//! Single-instance SQLite operations, confirmation holds, and in-memory +//! resource occupancy. Occupancy is not a SQLite schema. Request-owned +//! conflicts wait on a per-resource FIFO that starts at `acquire()`. A live +//! process rejects overlapping mutation immediately with `WORKSPACE_BUSY` +//! and fails trailing waiters. Queue saturation is `RESOURCE_QUEUE_FULL`. //! HTTP/JSON-RPC request ids are never stored as [`OperationId`] values. mod approvals; @@ -19,6 +23,7 @@ use sha2::{Digest, Sha256}; pub use approvals::{ApprovalRecord, ResumeClaim}; pub use coord::CreateIntent; +use resource::AcquireOutcome; pub use resource::{LockMode, Resource, ResourceGuard}; #[derive(Debug, Clone)] @@ -57,6 +62,84 @@ impl Drop for WriteGuard<'_> { } } +enum ReservationPhase { + PreDispatch, + Dispatching, + Confirmed, + Aborted, +} + +/// Workspace exclusive for an exec that has not necessarily spawned yet. +/// +/// Drop in `PreDispatch` releases the spawn reservation. After +/// `arm_dispatch()`, Drop does not release: Runner may already own a process. +pub struct ProcessReservation<'a> { + store: &'a Store, + process_id: String, + phase: ReservationPhase, +} + +impl ProcessReservation<'_> { + pub fn arm_dispatch(&mut self) { + if matches!(self.phase, ReservationPhase::PreDispatch) { + self.phase = ReservationPhase::Dispatching; + } + } + + pub fn confirm(mut self) { + self.store.confirm_process(&self.process_id); + self.phase = ReservationPhase::Confirmed; + } + + pub fn abort(mut self) { + self.store.release_process(&self.process_id); + self.phase = ReservationPhase::Aborted; + } +} + +impl Drop for ProcessReservation<'_> { + fn drop(&mut self) { + if matches!(self.phase, ReservationPhase::PreDispatch) { + self.store.release_process(&self.process_id); + } + } +} + +struct WaitTicket<'a> { + store: &'a Store, + resource: Resource, + mode: LockMode, + id: u64, + rx: tokio::sync::oneshot::Receiver>, + finished: bool, +} + +impl WaitTicket<'_> { + async fn wait(mut self) -> Result<(), ErrorBody> { + let result = match (&mut self.rx).await { + Ok(value) => value, + Err(_) => Err(waiter_closed_error()), + }; + self.finished = true; + result + } +} + +impl Drop for WaitTicket<'_> { + fn drop(&mut self) { + if self.finished { + return; + } + let mut serializer = self.store.locks.lock().expect("lock mutex"); + if serializer.cancel_waiter(&self.resource, self.id) { + return; + } + if matches!(self.rx.try_recv(), Ok(Ok(()))) { + serializer.unlock(&self.resource, self.mode); + } + } +} + impl Store { pub fn memory() -> Result { Self::from_connection(Connection::open_in_memory().map_err(|e| e.to_string())?) @@ -172,6 +255,24 @@ impl Store { }) } + pub async fn acquire_write<'a>( + &'a self, + workspace_id: &str, + ) -> Result, ErrorBody> { + let resource = Resource::Workspace(workspace_id.to_string()); + let outcome = self + .locks + .lock() + .expect("lock mutex") + .acquire_exclusive_write(workspace_id); + self.complete_acquire(resource, LockMode::Exclusive, outcome) + .await?; + Ok(WriteGuard { + store: self, + workspace_id: workspace_id.to_string(), + }) + } + fn release_write(&self, workspace_id: &str) { self.locks .lock() @@ -186,6 +287,33 @@ impl Store { .mark_shell_busy(workspace_id, process_id) } + pub async fn acquire_shell_busy( + &self, + workspace_id: &str, + process_id: &str, + ) -> Result, ErrorBody> { + let resource = Resource::Workspace(workspace_id.to_string()); + let outcome = self + .locks + .lock() + .expect("lock mutex") + .acquire_shell_busy(workspace_id, process_id); + self.complete_acquire(resource, LockMode::Exclusive, outcome) + .await?; + Ok(ProcessReservation { + store: self, + process_id: process_id.to_string(), + phase: ReservationPhase::PreDispatch, + }) + } + + fn confirm_process(&self, process_id: &str) { + self.locks + .lock() + .expect("lock mutex") + .confirm_process(process_id); + } + pub fn clear_shell(&self, workspace_id: &str) { self.locks .lock() @@ -219,6 +347,21 @@ impl Store { Ok(ResourceGuard::new(self, resource, mode)) } + pub async fn lock( + &self, + resource: Resource, + mode: LockMode, + ) -> Result, ErrorBody> { + let outcome = self + .locks + .lock() + .expect("lock mutex") + .acquire_lock(resource.clone(), mode); + self.complete_acquire(resource.clone(), mode, outcome) + .await?; + Ok(ResourceGuard::new(self, resource, mode)) + } + fn release_resource(&self, resource: &Resource, mode: LockMode) { self.locks .lock() @@ -226,6 +369,46 @@ impl Store { .unlock(resource, mode); } + async fn complete_acquire( + &self, + resource: Resource, + mode: LockMode, + outcome: AcquireOutcome, + ) -> Result<(), ErrorBody> { + match outcome { + AcquireOutcome::Granted => Ok(()), + AcquireOutcome::Busy(err) => Err(err), + AcquireOutcome::Waiting { id, rx } => { + WaitTicket { + store: self, + resource, + mode, + id, + rx, + finished: false, + } + .wait() + .await + } + } + } + + #[cfg(test)] + pub(crate) fn waiter_count(&self, resource: &Resource) -> usize { + self.locks + .lock() + .expect("lock mutex") + .waiter_count(resource) + } + + #[cfg(test)] + pub(crate) fn exclusive_kind(&self, resource: &Resource) -> Option<&'static str> { + self.locks + .lock() + .expect("lock mutex") + .exclusive_kind(resource) + } + pub fn begin( &self, operation_key: Option<&OperationKey>, @@ -477,9 +660,10 @@ fn migrate_approvals(conn: &Connection) -> Result<(), String> { .map_err(|err| err.to_string())?; } conn.execute_batch( - "CREATE UNIQUE INDEX IF NOT EXISTS idx_approvals_active_fp + "DROP INDEX IF EXISTS idx_approvals_active_fp; + CREATE UNIQUE INDEX IF NOT EXISTS idx_approvals_active_fp ON approvals(workspace_id, fingerprint) - WHERE state IN ('pending','granted','resuming') AND length(fingerprint) > 0;", + WHERE state IN ('pending','granted','queued','resuming') AND length(fingerprint) > 0;", ) .map_err(|err| err.to_string())?; Ok(()) @@ -514,6 +698,10 @@ fn status_str(status: PatchStatus) -> &'static str { } } +fn waiter_closed_error() -> ErrorBody { + ErrorBody::new(ErrorCode::Internal, "resource waiter closed") +} + fn sql_err(err: rusqlite::Error) -> ErrorBody { ErrorBody::new(ErrorCode::OperationNotFound, err.to_string()) } @@ -841,4 +1029,24 @@ mod tests { assert_eq!(status.events[0].name, OperationEventName::Minted); assert_eq!(status.events[1].name, OperationEventName::Finished); } + + #[tokio::test] + async fn waiter_closed_is_not_workspace_busy() { + let store = Store::memory().unwrap(); + let (tx, rx) = tokio::sync::oneshot::channel(); + drop(tx); + let err = WaitTicket { + store: &store, + resource: Resource::Workspace("demo".into()), + mode: LockMode::Exclusive, + id: 0, + rx, + finished: false, + } + .wait() + .await + .unwrap_err(); + assert_eq!(err.code, ErrorCode::Internal); + assert_eq!(err.message, "resource waiter closed"); + } } diff --git a/crates/store/src/resource.rs b/crates/store/src/resource.rs index 76191ec..50d455b 100644 --- a/crates/store/src/resource.rs +++ b/crates/store/src/resource.rs @@ -1,8 +1,20 @@ //! In-memory resource serialization. Not a SQLite schema and not a thread queue. +//! +//! FIFO order starts when an eligible request reaches `acquire()`, not when +//! the MCP message arrives. Request-owned conflicts wait. A confirmed live +//! process is a barrier: new acquires fail immediately and trailing waiters +//! are closed with `WORKSPACE_BUSY`. Spawn reservation (`Spawning`) is not +//! that barrier; spawn failure releases and wakes the next waiter. +//! +//! Live MCP mutations take only `Resource::Workspace`. Before any code takes +//! two resources at once, define a canonical order or `acquire_many`. Path +//! occupancy is typed but unused. `busy()` still maps unused resource kinds +//! to `WORKSPACE_BUSY`; split that taxonomy before those keys are public. -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; -use codespace_domain::{ErrorBody, ErrorCode}; +use codespace_domain::{ErrorBody, ErrorCode, MAX_WAITERS_PER_RESOURCE}; +use tokio::sync::oneshot; #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub enum Resource { @@ -25,35 +37,64 @@ pub enum LockMode { enum ExclusiveHolder { /// Request-owned RAII exclusive (`WriteGuard`). Request, - /// Process-owned exclusive (`ProcessId`). + /// Exec granted the lease but Runner spawn has not been confirmed. + Spawning(String), + /// Confirmed or unknown live process (`ProcessId`). Process(String), } +struct Waiter { + id: u64, + mode: LockMode, + owner: ExclusiveHolder, + tx: oneshot::Sender>, +} + #[derive(Default)] pub(crate) struct ResourceSerializer { exclusive: HashMap, shared: HashMap, + waiters: HashMap>, + next_waiter: u64, +} + +pub(crate) enum AcquireOutcome { + Granted, + Busy(ErrorBody), + Waiting { + id: u64, + rx: oneshot::Receiver>, + }, } impl ResourceSerializer { pub(crate) fn try_exclusive_write(&mut self, workspace_id: &str) -> Result<(), ErrorBody> { let resource = Resource::Workspace(workspace_id.to_string()); - if let Some(ExclusiveHolder::Process(_)) = self.exclusive.get(&resource) { - return Err(ErrorBody::new( - ErrorCode::WorkspaceBusy, - "workspace has a busy shell", - )); + self.wake(&resource); + if self.process_held(&resource) { + return Err(shell_busy()); } - if self.exclusive.contains_key(&resource) || self.shared_count(&resource) > 0 { + if self.exclusive.contains_key(&resource) + || self.shared_count(&resource) > 0 + || self.has_waiters(&resource) + { return Err(ErrorBody::new( ErrorCode::WorkspaceBusy, "workspace write lock is held", )); } - self.exclusive.insert(resource, ExclusiveHolder::Request); + self.grant(resource, LockMode::Exclusive, ExclusiveHolder::Request); Ok(()) } + pub(crate) fn acquire_exclusive_write(&mut self, workspace_id: &str) -> AcquireOutcome { + self.acquire( + Resource::Workspace(workspace_id.to_string()), + LockMode::Exclusive, + ExclusiveHolder::Request, + ) + } + pub(crate) fn release_write(&mut self, workspace_id: &str) { let resource = Resource::Workspace(workspace_id.to_string()); if matches!( @@ -61,6 +102,7 @@ impl ResourceSerializer { Some(ExclusiveHolder::Request) ) { self.exclusive.remove(&resource); + self.wake(&resource); } } @@ -70,17 +112,49 @@ impl ResourceSerializer { process_id: &str, ) -> Result<(), ErrorBody> { let resource = Resource::Workspace(workspace_id.to_string()); - if self.exclusive.contains_key(&resource) || self.shared_count(&resource) > 0 { + self.wake(&resource); + if self.exclusive.contains_key(&resource) + || self.shared_count(&resource) > 0 + || self.has_waiters(&resource) + { return Err(ErrorBody::new( ErrorCode::WorkspaceBusy, "workspace write lock is held", )); } - self.exclusive - .insert(resource, ExclusiveHolder::Process(process_id.to_string())); + self.grant( + resource, + LockMode::Exclusive, + ExclusiveHolder::Process(process_id.to_string()), + ); Ok(()) } + pub(crate) fn acquire_shell_busy( + &mut self, + workspace_id: &str, + process_id: &str, + ) -> AcquireOutcome { + self.acquire( + Resource::Workspace(workspace_id.to_string()), + LockMode::Exclusive, + ExclusiveHolder::Spawning(process_id.to_string()), + ) + } + + pub(crate) fn confirm_process(&mut self, process_id: &str) { + let mut targets = Vec::new(); + for (resource, holder) in &mut self.exclusive { + if matches!(holder, ExclusiveHolder::Spawning(id) if id == process_id) { + *holder = ExclusiveHolder::Process(process_id.to_string()); + targets.push(resource.clone()); + } + } + for resource in targets { + self.fail_waiters(&resource, shell_busy()); + } + } + pub(crate) fn clear_shell(&mut self, workspace_id: &str) { let resource = Resource::Workspace(workspace_id.to_string()); if matches!( @@ -88,43 +162,180 @@ impl ResourceSerializer { Some(ExclusiveHolder::Process(_)) ) { self.exclusive.remove(&resource); + self.wake(&resource); } } pub(crate) fn release_process(&mut self, process_id: &str) { - self.exclusive.retain(|_, holder| match holder { - ExclusiveHolder::Process(id) => id != process_id, - ExclusiveHolder::Request => true, + let released = self.release_matching(|holder| match holder { + ExclusiveHolder::Spawning(id) | ExclusiveHolder::Process(id) => id == process_id, + ExclusiveHolder::Request => false, }); + for resource in released { + self.wake(&resource); + } } pub(crate) fn release_all_processes(&mut self) { - self.exclusive.retain(|_, holder| match holder { - ExclusiveHolder::Process(_) => false, - ExclusiveHolder::Request => true, + let released = self.release_matching(|holder| { + matches!( + holder, + ExclusiveHolder::Spawning(_) | ExclusiveHolder::Process(_) + ) }); + for resource in released { + self.wake(&resource); + } } pub(crate) fn try_lock(&mut self, resource: Resource, mode: LockMode) -> Result<(), ErrorBody> { + self.wake(&resource); + if !self.can_try_grant(&resource, mode) { + return Err(busy(&resource)); + } + self.grant(resource, mode, ExclusiveHolder::Request); + Ok(()) + } + + pub(crate) fn acquire_lock(&mut self, resource: Resource, mode: LockMode) -> AcquireOutcome { + self.acquire(resource, mode, ExclusiveHolder::Request) + } + + pub(crate) fn unlock(&mut self, resource: &Resource, mode: LockMode) { match mode { LockMode::SharedRead => { - if self.exclusive.contains_key(&resource) { - return Err(busy(&resource)); + if let Some(count) = self.shared.get_mut(resource) { + *count = count.saturating_sub(1); + if *count == 0 { + self.shared.remove(resource); + } } - *self.shared.entry(resource).or_insert(0) += 1; - Ok(()) } LockMode::Exclusive => { - if self.exclusive.contains_key(&resource) || self.shared_count(&resource) > 0 { - return Err(busy(&resource)); - } - self.exclusive.insert(resource, ExclusiveHolder::Request); - Ok(()) + self.exclusive.remove(resource); } } + self.wake(resource); } - pub(crate) fn unlock(&mut self, resource: &Resource, mode: LockMode) { + pub(crate) fn cancel_waiter(&mut self, resource: &Resource, id: u64) -> bool { + let Some(queue) = self.waiters.get_mut(resource) else { + return false; + }; + let before = queue.len(); + queue.retain(|waiter| waiter.id != id); + let removed = queue.len() != before; + if queue.is_empty() { + self.waiters.remove(resource); + } + if removed { + self.wake(resource); + } + removed + } + + #[cfg(test)] + pub(crate) fn waiter_count(&self, resource: &Resource) -> usize { + self.waiters.get(resource).map(VecDeque::len).unwrap_or(0) + } + + #[cfg(test)] + pub(crate) fn exclusive_kind(&self, resource: &Resource) -> Option<&'static str> { + match self.exclusive.get(resource) { + Some(ExclusiveHolder::Request) => Some("request"), + Some(ExclusiveHolder::Spawning(_)) => Some("spawning"), + Some(ExclusiveHolder::Process(_)) => Some("process"), + None => None, + } + } + + fn acquire( + &mut self, + resource: Resource, + mode: LockMode, + owner: ExclusiveHolder, + ) -> AcquireOutcome { + if self.process_held(&resource) { + return AcquireOutcome::Busy(process_held_busy(&resource)); + } + let fairness_blocks = mode == LockMode::SharedRead && self.has_exclusive_waiter(&resource); + if self.can_grant(&resource, mode) + && !fairness_blocks + && (mode == LockMode::SharedRead || !self.has_waiters(&resource)) + { + self.grant(resource, mode, owner); + return AcquireOutcome::Granted; + } + if self.waiter_count_unlocked(&resource) >= MAX_WAITERS_PER_RESOURCE as usize { + return AcquireOutcome::Busy(queue_full()); + } + let (id, rx) = self.enqueue(resource.clone(), mode, owner); + self.wake(&resource); + AcquireOutcome::Waiting { id, rx } + } + + fn fail_waiters(&mut self, resource: &Resource, err: ErrorBody) { + let Some(queue) = self.waiters.remove(resource) else { + return; + }; + for waiter in queue { + let _ = waiter.tx.send(Err(err.clone())); + } + } + + fn enqueue( + &mut self, + resource: Resource, + mode: LockMode, + owner: ExclusiveHolder, + ) -> (u64, oneshot::Receiver>) { + let id = self.next_waiter; + self.next_waiter = self.next_waiter.wrapping_add(1); + let (tx, rx) = oneshot::channel(); + self.waiters.entry(resource).or_default().push_back(Waiter { + id, + mode, + owner, + tx, + }); + (id, rx) + } + + fn wake(&mut self, resource: &Resource) { + loop { + let Some(front_mode) = self + .waiters + .get(resource) + .and_then(|queue| queue.front().map(|waiter| waiter.mode)) + else { + self.waiters.remove(resource); + return; + }; + if !self.can_grant(resource, front_mode) { + return; + } + let Some(waiter) = self.waiters.get_mut(resource).and_then(VecDeque::pop_front) else { + return; + }; + if self + .waiters + .get(resource) + .is_none_or(|queue| queue.is_empty()) + { + self.waiters.remove(resource); + } + self.grant(resource.clone(), waiter.mode, waiter.owner); + if waiter.tx.send(Ok(())).is_err() { + self.unlock_silent(resource, waiter.mode); + continue; + } + if waiter.mode == LockMode::Exclusive { + return; + } + } + } + + fn unlock_silent(&mut self, resource: &Resource, mode: LockMode) { match mode { LockMode::SharedRead => { if let Some(count) = self.shared.get_mut(resource) { @@ -140,12 +351,84 @@ impl ResourceSerializer { } } + fn grant(&mut self, resource: Resource, mode: LockMode, owner: ExclusiveHolder) { + match mode { + LockMode::SharedRead => { + *self.shared.entry(resource).or_insert(0) += 1; + } + LockMode::Exclusive => { + self.exclusive.insert(resource, owner); + } + } + } + + fn can_grant(&self, resource: &Resource, mode: LockMode) -> bool { + match mode { + LockMode::SharedRead => !self.exclusive.contains_key(resource), + LockMode::Exclusive => { + !self.exclusive.contains_key(resource) && self.shared_count(resource) == 0 + } + } + } + + fn can_try_grant(&self, resource: &Resource, mode: LockMode) -> bool { + match mode { + LockMode::SharedRead => { + self.can_grant(resource, mode) && !self.has_exclusive_waiter(resource) + } + LockMode::Exclusive => self.can_grant(resource, mode) && !self.has_waiters(resource), + } + } + + fn process_held(&self, resource: &Resource) -> bool { + matches!(resource, Resource::Workspace(_)) + && matches!( + self.exclusive.get(resource), + Some(ExclusiveHolder::Process(_)) + ) + } + + fn waiter_count_unlocked(&self, resource: &Resource) -> usize { + self.waiters.get(resource).map(VecDeque::len).unwrap_or(0) + } + + fn has_waiters(&self, resource: &Resource) -> bool { + self.waiter_count_unlocked(resource) > 0 + } + + fn has_exclusive_waiter(&self, resource: &Resource) -> bool { + self.waiters.get(resource).is_some_and(|queue| { + queue + .iter() + .any(|waiter| waiter.mode == LockMode::Exclusive) + }) + } + fn shared_count(&self, resource: &Resource) -> u32 { self.shared.get(resource).copied().unwrap_or(0) } + + fn release_matching( + &mut self, + mut pred: impl FnMut(&ExclusiveHolder) -> bool, + ) -> Vec { + let mut released = Vec::new(); + self.exclusive.retain(|resource, holder| { + if pred(holder) { + released.push(resource.clone()); + false + } else { + true + } + }); + released + } } fn busy(resource: &Resource) -> ErrorBody { + // Unused resource kinds are not a public occupancy contract yet. + // Split this taxonomy before Environment/Path/Process/Operation/Watch + // become live MCP locks. let detail = match resource { Resource::Workspace(_) => "workspace write lock is held", Resource::Environment(_) => "environment lock is held", @@ -157,6 +440,24 @@ fn busy(resource: &Resource) -> ErrorBody { ErrorBody::new(ErrorCode::WorkspaceBusy, detail) } +fn shell_busy() -> ErrorBody { + ErrorBody::new(ErrorCode::WorkspaceBusy, "workspace has a busy shell") +} + +fn queue_full() -> ErrorBody { + ErrorBody::new( + ErrorCode::ResourceQueueFull, + format!("resource waiter queue exceeds {MAX_WAITERS_PER_RESOURCE}"), + ) +} + +fn process_held_busy(resource: &Resource) -> ErrorBody { + match resource { + Resource::Workspace(_) => shell_busy(), + _ => busy(resource), + } +} + pub struct ResourceGuard<'a> { store: &'a super::Store, resource: Resource, @@ -183,6 +484,7 @@ impl Drop for ResourceGuard<'_> { mod tests { use super::*; use crate::Store; + use std::sync::{Arc, Mutex}; #[test] fn shared_read_does_not_take_exclusive() { @@ -237,4 +539,342 @@ mod tests { .try_lock(Resource::Workspace("demo".into()), LockMode::SharedRead) .is_err()); } + + async fn wait_for_waiters(store: &Store, resource: &Resource, n: usize) { + for _ in 0..200 { + if store.waiter_count(resource) >= n { + return; + } + tokio::task::yield_now().await; + std::thread::sleep(std::time::Duration::from_millis(1)); + } + panic!("expected {n} waiters, got {}", store.waiter_count(resource)); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn request_exclusive_fifo_order() { + let store = Arc::new(Store::memory().unwrap()); + let resource = Resource::Workspace("demo".into()); + let first = store.acquire_write("demo").await.unwrap(); + let order = Arc::new(Mutex::new(Vec::new())); + + let store_b = store.clone(); + let order_b = order.clone(); + let b = tokio::spawn(async move { + let guard = store_b.acquire_write("demo").await.unwrap(); + order_b.lock().expect("order").push("b"); + drop(guard); + }); + wait_for_waiters(&store, &resource, 1).await; + + let store_c = store.clone(); + let order_c = order.clone(); + let c = tokio::spawn(async move { + let guard = store_c.acquire_write("demo").await.unwrap(); + order_c.lock().expect("order").push("c"); + drop(guard); + }); + wait_for_waiters(&store, &resource, 2).await; + + drop(first); + b.await.unwrap(); + c.await.unwrap(); + assert_eq!(*order.lock().expect("order"), vec!["b", "c"]); + } + + #[tokio::test] + async fn process_held_workspace_rejects_without_queueing() { + let store = Store::memory().unwrap(); + store.mark_shell_busy("demo", "proc-1").unwrap(); + let err = match store.acquire_write("demo").await { + Err(err) => err, + Ok(_) => panic!("process-held write should be busy"), + }; + assert_eq!(err.code, ErrorCode::WorkspaceBusy); + assert_eq!(store.waiter_count(&Resource::Workspace("demo".into())), 0); + let err = match store + .lock(Resource::Workspace("demo".into()), LockMode::SharedRead) + .await + { + Err(err) => err, + Ok(_) => panic!("process-held shared read should be busy"), + }; + assert_eq!(err.code, ErrorCode::WorkspaceBusy); + assert_eq!(store.waiter_count(&Resource::Workspace("demo".into())), 0); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn cancelled_waiter_does_not_block_next_grant() { + let store = Arc::new(Store::memory().unwrap()); + let resource = Resource::Workspace("demo".into()); + let first = store.acquire_write("demo").await.unwrap(); + let store_b = store.clone(); + let b = tokio::spawn(async move { + store_b.acquire_write("demo").await.unwrap(); + }); + wait_for_waiters(&store, &resource, 1).await; + b.abort(); + let _ = b.await; + for _ in 0..50 { + if store.waiter_count(&resource) == 0 { + break; + } + tokio::task::yield_now().await; + std::thread::sleep(std::time::Duration::from_millis(1)); + } + assert_eq!(store.waiter_count(&resource), 0); + drop(first); + let _next = store.acquire_write("demo").await.unwrap(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn shared_read_waits_behind_request_exclusive() { + let store = Arc::new(Store::memory().unwrap()); + let resource = Resource::Path { + workspace: "demo".into(), + path: "a.txt".into(), + }; + let exclusive = store + .lock(resource.clone(), LockMode::Exclusive) + .await + .unwrap(); + let store_r = store.clone(); + let resource_r = resource.clone(); + let (held_tx, held_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let reader = tokio::spawn(async move { + let _shared = store_r + .lock(resource_r, LockMode::SharedRead) + .await + .unwrap(); + held_tx.send(()).expect("held"); + let _ = release_rx.await; + }); + wait_for_waiters(&store, &resource, 1).await; + drop(exclusive); + held_rx.await.expect("shared granted"); + assert!(store + .try_lock(resource.clone(), LockMode::Exclusive) + .is_err()); + release_tx.send(()).expect("release"); + reader.await.unwrap(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn shared_read_does_not_barge_ahead_of_exclusive_waiter() { + let store = Arc::new(Store::memory().unwrap()); + let resource = Resource::Path { + workspace: "demo".into(), + path: "a.txt".into(), + }; + let first = store + .lock(resource.clone(), LockMode::Exclusive) + .await + .unwrap(); + let order = Arc::new(Mutex::new(Vec::new())); + + let store_ex = store.clone(); + let resource_ex = resource.clone(); + let order_ex = order.clone(); + let (ex_held_tx, ex_held_rx) = tokio::sync::oneshot::channel(); + let (ex_release_tx, ex_release_rx) = tokio::sync::oneshot::channel(); + let exclusive_waiter = tokio::spawn(async move { + let _guard = store_ex + .lock(resource_ex, LockMode::Exclusive) + .await + .unwrap(); + order_ex.lock().expect("order").push("exclusive"); + ex_held_tx.send(()).expect("held"); + let _ = ex_release_rx.await; + }); + wait_for_waiters(&store, &resource, 1).await; + + let store_sh = store.clone(); + let resource_sh = resource.clone(); + let order_sh = order.clone(); + let (sh_held_tx, sh_held_rx) = tokio::sync::oneshot::channel(); + let (sh_release_tx, sh_release_rx) = tokio::sync::oneshot::channel(); + let shared_waiter = tokio::spawn(async move { + let _guard = store_sh + .lock(resource_sh, LockMode::SharedRead) + .await + .unwrap(); + order_sh.lock().expect("order").push("shared"); + sh_held_tx.send(()).expect("held"); + let _ = sh_release_rx.await; + }); + wait_for_waiters(&store, &resource, 2).await; + + drop(first); + ex_held_rx.await.expect("exclusive granted"); + assert_eq!(*order.lock().expect("order"), vec!["exclusive"]); + ex_release_tx.send(()).expect("release exclusive"); + exclusive_waiter.await.unwrap(); + sh_held_rx.await.expect("shared granted"); + sh_release_tx.send(()).expect("release shared"); + shared_waiter.await.unwrap(); + assert_eq!(*order.lock().expect("order"), vec!["exclusive", "shared"]); + } + + #[tokio::test] + async fn workspaces_are_independent() { + let store = Store::memory().unwrap(); + let _demo = store.acquire_write("demo").await.unwrap(); + let _other = store.acquire_write("other").await.unwrap(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn exec_waits_behind_request_owned_write() { + let store = Arc::new(Store::memory().unwrap()); + let resource = Resource::Workspace("demo".into()); + let write = store.acquire_write("demo").await.unwrap(); + let store_exec = store.clone(); + let exec = tokio::spawn(async move { + let reservation = store_exec.acquire_shell_busy("demo", "proc-wait").await?; + reservation.confirm(); + Ok::<(), ErrorBody>(()) + }); + wait_for_waiters(&store, &resource, 1).await; + drop(write); + exec.await.unwrap().unwrap(); + assert_eq!(store.exclusive_kind(&resource), Some("process")); + assert_eq!( + store.try_acquire_write("demo").err().map(|e| e.code), + Some(ErrorCode::WorkspaceBusy) + ); + store.release_process("proc-wait"); + let _next = store.try_acquire_write("demo").unwrap(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn confirm_live_process_fails_trailing_waiters() { + let store = Arc::new(Store::memory().unwrap()); + let resource = Resource::Workspace("demo".into()); + let write = store.acquire_write("demo").await.unwrap(); + let store_exec = store.clone(); + let exec = tokio::spawn(async move { + let reservation = store_exec.acquire_shell_busy("demo", "proc-live").await?; + reservation.confirm(); + Ok::<(), ErrorBody>(()) + }); + wait_for_waiters(&store, &resource, 1).await; + let store_patch = store.clone(); + let trailing = tokio::spawn(async move { + match store_patch.acquire_write("demo").await { + Ok(_) => panic!("trailing waiter should be busy"), + Err(err) => err, + } + }); + wait_for_waiters(&store, &resource, 2).await; + drop(write); + exec.await.unwrap().unwrap(); + let err = trailing.await.unwrap(); + assert_eq!(err.code, ErrorCode::WorkspaceBusy); + assert_eq!(store.waiter_count(&resource), 0); + let late = match store.acquire_write("demo").await { + Err(err) => err, + Ok(_) => panic!("late write should be busy"), + }; + assert_eq!(late.code, ErrorCode::WorkspaceBusy); + store.release_process("proc-live"); + let _next = store.acquire_write("demo").await.unwrap(); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn spawn_abort_wakes_next_waiter() { + let store = Arc::new(Store::memory().unwrap()); + let resource = Resource::Workspace("demo".into()); + let write = store.acquire_write("demo").await.unwrap(); + let store_exec = store.clone(); + let exec = tokio::spawn(async move { + let reservation = store_exec.acquire_shell_busy("demo", "proc-fail").await?; + reservation.abort(); + Ok::<(), ErrorBody>(()) + }); + wait_for_waiters(&store, &resource, 1).await; + let store_patch = store.clone(); + let next = tokio::spawn(async move { + let _guard = store_patch.acquire_write("demo").await?; + Ok::<(), ErrorBody>(()) + }); + wait_for_waiters(&store, &resource, 2).await; + drop(write); + exec.await.unwrap().unwrap(); + next.await.unwrap().unwrap(); + assert_eq!(store.exclusive_kind(&resource), None); + } + + #[tokio::test] + async fn pre_dispatch_drop_releases_spawn_reservation() { + let store = Store::memory().unwrap(); + let resource = Resource::Workspace("demo".into()); + { + let _reservation = store.acquire_shell_busy("demo", "proc-drop").await.unwrap(); + assert_eq!(store.exclusive_kind(&resource), Some("spawning")); + } + assert_eq!(store.exclusive_kind(&resource), None); + let _write = store.acquire_write("demo").await.unwrap(); + } + + #[tokio::test] + async fn dispatching_drop_keeps_spawn_reservation() { + let store = Store::memory().unwrap(); + let resource = Resource::Workspace("demo".into()); + { + let mut reservation = store.acquire_shell_busy("demo", "proc-arm").await.unwrap(); + reservation.arm_dispatch(); + } + assert_eq!(store.exclusive_kind(&resource), Some("spawning")); + store.release_process("proc-arm"); + assert_eq!(store.exclusive_kind(&resource), None); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn queue_depth_bound_rejects_without_enqueue() { + let store = Arc::new(Store::memory().unwrap()); + let resource = Resource::Workspace("demo".into()); + let first = store.acquire_write("demo").await.unwrap(); + let mut waiters = Vec::new(); + for _ in 0..MAX_WAITERS_PER_RESOURCE { + let store_w = store.clone(); + waiters.push(tokio::spawn(async move { + let _guard = store_w.acquire_write("demo").await?; + std::future::pending::<()>().await; + Ok::<(), ErrorBody>(()) + })); + } + wait_for_waiters(&store, &resource, MAX_WAITERS_PER_RESOURCE as usize).await; + let err = match store.acquire_write("demo").await { + Err(err) => err, + Ok(_) => panic!("queue should be full"), + }; + assert_eq!(err.code, ErrorCode::ResourceQueueFull); + assert_eq!( + store.waiter_count(&resource), + MAX_WAITERS_PER_RESOURCE as usize + ); + waiters.pop().unwrap().abort(); + for _ in 0..50 { + if store.waiter_count(&resource) < MAX_WAITERS_PER_RESOURCE as usize { + break; + } + tokio::task::yield_now().await; + std::thread::sleep(std::time::Duration::from_millis(1)); + } + assert!(store.waiter_count(&resource) < MAX_WAITERS_PER_RESOURCE as usize); + let recovered = tokio::spawn({ + let store = store.clone(); + async move { + let _guard = store.acquire_write("demo").await?; + std::future::pending::<()>().await; + Ok::<(), ErrorBody>(()) + } + }); + wait_for_waiters(&store, &resource, MAX_WAITERS_PER_RESOURCE as usize).await; + for waiter in waiters { + waiter.abort(); + } + recovered.abort(); + drop(first); + } } diff --git a/docs/agent-integration.md b/docs/agent-integration.md index 4a82911..67d2e15 100644 --- a/docs/agent-integration.md +++ b/docs/agent-integration.md @@ -175,7 +175,7 @@ Default workspaces run allowed patches and commands immediately. If the operator } ``` -`approval_resolve` does not change the permission profile. `operation_resume` re-checks policy, then runs the original apply or exec path once. `consumed` is stored only with a terminal result. Repeating resume returns that stored result, recovers a recorded patch from the operations ledger, or returns `APPROVAL_AMBIGUOUS`. An interrupted exec resume is not respawned. Deny is terminal. The same three tools exist when approvals are `off`; only an explicit `approval_create` opens a hold in that mode. v1 does not distinguish host from model: the same MCP caller can grant. +`approval_resolve` does not change the permission profile. `operation_resume` claims `granted` into `queued`, re-checks policy, waits for the resource, then moves to `resuming` only when dispatch can start. Restart from `queued` reacquires. Restart from `resuming` returns a stored result, recovers a recorded patch, or returns `APPROVAL_AMBIGUOUS`. `consumed` is stored only with a terminal result. An interrupted exec resume is not respawned. Deny is terminal. The same three tools exist when approvals are `off`; only an explicit `approval_create` opens a hold in that mode. v1 does not distinguish host from model: the same MCP caller can grant. ## Retry and recover deliberately @@ -191,7 +191,8 @@ Default workspaces run allowed patches and commands immediately. If the operator | Exec `dispatch_status: unknown` | A process may exist. Inspect or terminate the returned handle if reachable; do not blindly start another | | Lost spawn response / no `process_id` | Do not invent or search for a handle. Do not start a duplicate; the process may still occupy the workspace | | Client/HTTP session lost after spawn | Process keeps running. Reconnect and use the saved `process_id` | -| `WORKSPACE_BUSY` | Wait for the owning task or cancel the process; avoid a tight retry loop | +| `WORKSPACE_BUSY` | Live process owns the workspace, including waiters already queued behind that process. Wait or cancel the process. Avoid a tight retry loop | +| `RESOURCE_QUEUE_FULL` | Too many request-owned waiters; back off and retry later. Distinct from `WORKSPACE_BUSY` | | `TIMEOUT` | Treat execution as interrupted; inspect partial effects | | Server/worker lost | Reconnect and inspect capabilities/files; old process handles are not recoverable | diff --git a/docs/architecture.md b/docs/architecture.md index 15c443c..3345675 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -56,7 +56,7 @@ MCP request completion does not end a managed process. Clients continue with its ## Extension boundaries -The core does not import Codex types directly. Adapters may depend on a broader Codex execution graph; this does not make the gateway a Codex agent. Operator configuration selects environments, while MCP clients select only registered workspaces. Container execution, remote runners, and a resource scheduler are not implemented. +The core does not import Codex types directly. Adapters may depend on a broader Codex execution graph; this does not make the gateway a Codex agent. Operator configuration selects environments, while MCP clients select only registered workspaces. Container execution and remote runners are not implemented. Workspace occupancy waits on an in-memory per-resource FIFO for request-owned work. FIFO order starts at acquire. A live process returns `WORKSPACE_BUSY` for trailing waiters and new arrivals. Queue saturation is `RESOURCE_QUEUE_FULL`. Confirmation-hold tools (`approval_create`, `approval_resolve`, `operation_resume`) are implemented. They pause a mutation the profile already allows until the hold is granted. This is not a security boundary: they do not raise `read-only` to write/exec, honor `ClientClaims.approved`, or change the permission profile. The same MCP caller can grant. Resume re-checks policy. v1 does not separate host and model callers. diff --git a/docs/behavior-differences.md b/docs/behavior-differences.md index 4a5cad3..8eb19e2 100644 --- a/docs/behavior-differences.md +++ b/docs/behavior-differences.md @@ -39,6 +39,6 @@ A successful result includes affected `files` and `changes` with kind and availa ## Write lock and transport -One live command blocks other patch/exec work in that workspace with `WORKSPACE_BUSY`. There is no waiting queue. Reads and searches remain possible. Patch operation keys support replay only for matching request fingerprints; choosing a new key after an uncertain response risks applying the change twice. +One live command blocks other patch/exec work in that workspace with `WORKSPACE_BUSY`, including waiters already queued behind that process. Request-owned patch and exec work waits in an in-memory FIFO that starts at acquire until the current request finishes. Queue saturation is `RESOURCE_QUEUE_FULL`. There is no durable occupancy queue. Reads and searches remain possible. Patch operation keys support replay only for matching request fingerprints; choosing a new key after an uncertain response risks applying the change twice. stdio and Streamable HTTP expose the same tool schemas. Connection failure does not establish whether a mutation ran. See [error codes](error-codes.md) and [recovery rules](agent-integration.md). diff --git a/docs/error-codes.md b/docs/error-codes.md index 13a5f58..95e0d36 100644 --- a/docs/error-codes.md +++ b/docs/error-codes.md @@ -22,7 +22,8 @@ Errors use uppercase identifiers and a message. An `operation_id` may be present | --- | --- | | `UNAUTHORIZED` | Action denied by tool policy | | `WORKSPACE_NOT_FOUND` | Workspace ID is not registered | -| `WORKSPACE_BUSY` | Another mutation or live command owns the workspace | +| `WORKSPACE_BUSY` | A live process owns the workspace mutation lease | +| `RESOURCE_QUEUE_FULL` | Too many request-owned waiters on one resource | | `INVALID_PATCH` | Patch parsing, preflight, or result verification failed; also retained in some helper/input error paths | | `INVALID_COMMAND` | Malformed argv rejected before dispatch | | `PROCESS_SPAWN_FAILED` | Backend confirmed no managed process was established | @@ -53,6 +54,7 @@ Errors use uppercase identifiers and a message. An `operation_id` may be present | `APPROVAL_NOT_FOUND` | Unknown confirmation-hold id | | `APPROVAL_CONFLICT` | Hold is still pending, already decided, or a resume is already in progress | | `APPROVAL_AMBIGUOUS` | Resume was interrupted and the terminal result is not on disk. Includes `approval_id`; patch cases may also include `operation_id` | +| `INTERNAL` | Scheduler or store invariant failed. Not occupancy; do not retry as `WORKSPACE_BUSY` | ## Dispatch and completion diff --git a/docs/execution-substrate.md b/docs/execution-substrate.md index 8ae46c1..91e9e27 100644 --- a/docs/execution-substrate.md +++ b/docs/execution-substrate.md @@ -26,7 +26,7 @@ The operator registry maps `read-only` and `workspace-write` to effective permis | Permission profile | Gateway-owned meaning of allowed file and process actions | | Operation | Persisted patch ledger with `operation_id`, optional idempotency key, `files`/`changes` hashes, and minted/finished events. Look up with `operation_status`. Does not track exec | | Process | Server-issued handle for a command; memory-only | -| Confirmation hold | Operator `approvals` setting; row in the `approvals` table (`pending`/`granted`/`resuming` until a terminal result); not a patch operation, privilege grant, or isolation boundary | +| Confirmation hold | Operator `approvals` setting; row in the `approvals` table (`pending`/`granted`/`queued`/`resuming` until a terminal result); not a patch operation, privilege grant, or isolation boundary | | Work | Logical job and user-instruction queue; separate from a transport session | `environment_id` is not an MCP tool argument. A network proxy URL or Codex user configuration supplied by the model does not become execution authority. @@ -41,12 +41,15 @@ Gateway fills workspace-root cwd, runner-local environment defaults, time/output `exec_command` returns dispatch identity (`process_id`, `dispatch_status`). `process_status` reports `running` or `exited` plus termination metadata. `process_resize` changes the size of a running PTY. `read_process` reports output including `output_lost` and `retained_from`. EOF is not success. After handle eviction the next lookup is `PROCESS_NOT_FOUND`, not a new state. Pipe-backed resize is `PROCESS_NOT_TTY`; an exited handle is `PROCESS_NOT_RUNNING`. On Linux sandbox, the wait status is that of the managed child (the helper argv); it is not documented as identical to the user argv. -A workspace mutation lease prevents simultaneous patch/exec mutations. Read and find remain available while a command runs, so filesystem I/O must reject symlink races at open time rather than rely on a prior path check. Runner file operations use `codespace-fs`; patch execution uses the separate patch helper. `operation_status` exposes the recorded patch ledger (`kind` is `patch`); live commands stay on `process_id` and are not recovered through that lookup. +A workspace mutation lease prevents simultaneous patch/exec mutations. FIFO order starts when an eligible request attempts to acquire the resource, not when the MCP message arrives. Request-owned patch and exec work on the same workspace waits in an in-memory per-resource FIFO until the current request releases the lease. When an exec becomes a live process, trailing waiters fail with `WORKSPACE_BUSY` and later arrivals are rejected immediately. Queue saturation is `RESOURCE_QUEUE_FULL`. That wait is not a durable or thread-keyed queue. Read and find remain available while a command runs, so filesystem I/O must reject symlink races at open time rather than rely on a prior path check. Runner file operations use `codespace-fs`; patch execution uses the separate patch helper. `operation_status` exposes the recorded patch ledger (`kind` is `patch`); live commands stay on `process_id` and are not recovered through that lookup. UDS transport and Linux sandbox preparation have distinct protocols and failure boundaries. A partially delivered UDS mutation may yield an uncertain result; never retry it as a new mutation merely because the connection failed. Process lifetime is owned by the runner instance: MCP/HTTP client disconnect keeps the process running, while UDS gateway↔worker loss or gateway shutdown terminates the owned subtree. There is no durable process recovery. The complete process and isolation rules belong in [runner isolation](runner-isolation.md). + +Occupancy uses an in-memory per-resource FIFO. FIFO order starts at `acquire()`. Request-owned exclusive waiters run in order when the current request drops the lease. A confirmed live process is a barrier: trailing waiters and new acquires return `WORKSPACE_BUSY`. Spawn reservation is not that barrier; spawn failure wakes the next waiter. Queue depth is bounded (`RESOURCE_QUEUE_FULL`). `read` and `find` do not take a shared lease. The queue is not stored in SQLite and is not keyed by thread. Live MCP mutations still use the workspace exclusive, not path-level locks. + @@ -65,11 +68,11 @@ The Runner may start a recursive filesystem watcher for a workspace on the first `approval_create`, `approval_resolve`, and `operation_resume` are callable. They hold a mutation the workspace profile already allows until the hold is granted. This is not a security boundary: they do not escalate permissions, apply `{ "network": true }` or `ClientClaims.approved`, or write V4A snapshots into the patch operations ledger. The same MCP caller can grant. -When the operator sets workspace `approvals` to `confirm`, a policy-allowed `apply_patch` or `exec_command` returns `APPROVAL_REQUIRED` with an `approval_id` before `begin()` or spawn. Retrying the same logical request reuses that active hold. `off` (the default) still runs those tools immediately; the three tools remain listed so an explicit `approval_create` can open a hold. Grant does not change the profile. Resume claims `granted` into `resuming`, re-checks `allow()`, then runs the existing apply or exec inner path. `consumed` is recorded only together with the terminal result. A later resume returns that result, recovers a patch from the operations ledger, or returns `APPROVAL_AMBIGUOUS`. Interrupted exec is not respawned. A denied policy stays `UNAUTHORIZED`. Exec remains on `process_id`. v1 does not authenticate host versus model. The server guarantees the policy re-check on resume plus that durability contract. +When the operator sets workspace `approvals` to `confirm`, a policy-allowed `apply_patch` or `exec_command` returns `APPROVAL_REQUIRED` with an `approval_id` before `begin()` or spawn. Retrying the same logical request reuses that active hold. `off` (the default) still runs those tools immediately; the three tools remain listed so an explicit `approval_create` can open a hold. Grant does not change the profile. Resume claims `granted` into `queued` (restart-safe; no side effect yet), re-checks `allow()`, waits for the resource, then moves to `resuming` only when dispatch can start. `consumed` is recorded only together with the terminal result. A later resume from `queued` reacquires. A later resume from `resuming` returns a stored result, recovers a patch from the operations ledger, or returns `APPROVAL_AMBIGUOUS`. Interrupted exec is not respawned. A denied policy stays `UNAUTHORIZED`. Exec remains on `process_id`. v1 does not authenticate host versus model. The server guarantees the policy re-check on resume plus that durability contract. ## What remains unimplemented -Durable process recovery, container/remote dispatch, and a resource queue scheduler remain absent. MCP `fs/watch`, UDS watch events, and write-cause classification are not provided. Richer internal types and negotiated protocol flags do not imply those features are callable. +Durable process recovery and container/remote dispatch remain absent. MCP `fs/watch`, UDS watch events, and write-cause classification are not provided. Richer internal types and negotiated protocol flags do not imply those features are callable. ## Maintaining the boundary diff --git a/docs/ko/agent-integration.md b/docs/ko/agent-integration.md index 88d7cbc..639939a 100644 --- a/docs/ko/agent-integration.md +++ b/docs/ko/agent-integration.md @@ -175,7 +175,7 @@ MCP 클라이언트 SDK로 stdio 또는 Streamable HTTP를 초기화하고, 초 } ``` -`approval_resolve`는 권한 프로필을 바꾸지 않습니다. `operation_resume`은 정책을 다시 검사한 뒤 원래 패치·실행 경로를 한 번 돌립니다. `consumed`는 단말 결과가 저장된 뒤에만 기록됩니다. 같은 홀드를 다시 재개하면 저장한 결과, 패치 원장 복구, 또는 `APPROVAL_AMBIGUOUS`가 반환됩니다. 중단된 exec 재개는 다시 spawn하지 않습니다. 거절은 단말입니다. `approvals`가 `off`여도 세 도구는 목록에 있으며, 그때는 명시적 `approval_create`만 홀드를 만듭니다. v1은 호스트와 모델을 구분하지 않으므로 같은 MCP 호출자가 grant할 수 있습니다. +`approval_resolve`는 권한 프로필을 바꾸지 않습니다. `operation_resume`은 `granted`를 `queued`로 옮긴 뒤 정책을 다시 검사하고, 자원을 받은 다음에야 `resuming`으로 올립니다. `queued`에서 재시작하면 다시 acquire합니다. `resuming`에서 재시작하면 저장한 결과, 패치 원장 복구, 또는 `APPROVAL_AMBIGUOUS`가 반환됩니다. `consumed`는 단말 결과가 저장된 뒤에만 기록됩니다. 중단된 exec 재개는 다시 spawn하지 않습니다. 거절은 단말입니다. `approvals`가 `off`여도 세 도구는 목록에 있으며, 그때는 명시적 `approval_create`만 홀드를 만듭니다. v1은 호스트와 모델을 구분하지 않으므로 같은 MCP 호출자가 grant할 수 있습니다. ## 재시도와 복구 @@ -191,7 +191,8 @@ MCP 클라이언트 SDK로 stdio 또는 Streamable HTTP를 초기화하고, 초 | 실행 `dispatch_status: unknown` | 프로세스가 존재할 수 있음. 연결 가능하면 해당 핸들을 조회·종료하고 무조건 재실행하지 않기 | | spawn 응답 유실 / `process_id` 없음 | 핸들을 만들어 내거나 검색하지 않기. 중복 시작하지 않기. 프로세스가 작업 공간을 점유 중일 수 있음 | | spawn 이후 클라이언트·HTTP 세션 끊김 | 프로세스는 계속 실행됨. 재연결 후 저장한 `process_id` 사용 | -| `WORKSPACE_BUSY` | 점유 중인 작업을 기다리거나 프로세스 취소. 빠른 반복 재시도 피하기 | +| `WORKSPACE_BUSY` | 라이브 프로세스가 작업 공간을 점유함. 이미 그 프로세스 뒤에 줄 선 대기자도 포함. 기다리거나 프로세스를 취소. 빠른 반복 재시도 피하기 | +| `RESOURCE_QUEUE_FULL` | 요청 소유 대기자가 한도 초과. 잠시 뒤 재시도. `WORKSPACE_BUSY`와는 다른 실패 | | `TIMEOUT` | 실행이 중단된 것으로 처리하고 일부 변경이 남았는지 확인 | | 서버·worker 연결 손실 | 재연결 후 기능과 파일 상태 확인. 기존 프로세스 핸들은 복구되지 않음 | diff --git a/docs/ko/architecture.md b/docs/ko/architecture.md index 163efa3..be3db1b 100644 --- a/docs/ko/architecture.md +++ b/docs/ko/architecture.md @@ -65,7 +65,7 @@ MCP 요청이 끝나도 관리 중인 프로세스는 유지됩니다. 클라이 ## 확장 시 유지할 경계 -핵심 계층은 Codex 타입을 직접 가져오지 않습니다. 어댑터가 여러 Codex 실행 구성 요소에 의존할 수는 있지만 게이트웨이가 Codex 에이전트가 되는 것은 아닙니다. 실행 환경은 운영자가 설정하고, MCP 클라이언트는 등록된 작업 공간만 선택합니다. 컨테이너 실행, 원격 러너, 자원 스케줄러는 아직 구현되지 않았습니다. +핵심 계층은 Codex 타입을 직접 가져오지 않습니다. 어댑터가 여러 Codex 실행 구성 요소에 의존할 수는 있지만 게이트웨이가 Codex 에이전트가 되는 것은 아닙니다. 실행 환경은 운영자가 설정하고, MCP 클라이언트는 등록된 작업 공간만 선택합니다. 컨테이너 실행과 원격 러너는 아직 구현되지 않았습니다. 작업 공간 점유는 요청 소유 작업에 대해 자원별 인메모리 FIFO에서 기다리며, FIFO는 `acquire()`에서 시작합니다. 라이브 프로세스는 뒤 대기자와 새 요청에 `WORKSPACE_BUSY`를 반환합니다. 큐 포화는 `RESOURCE_QUEUE_FULL`입니다. 확인 홀드 도구(`approval_create`, `approval_resolve`, `operation_resume`)는 구현되어 있습니다. 프로필이 이미 허용한 변경을 홀드가 승인될 때까지 멈춥니다. 보안 경계가 아닙니다. `read-only`를 쓰기·실행으로 올리거나 `ClientClaims.approved`를 인정하거나 프로필을 바꾸지 않습니다. 같은 MCP 호출자가 grant할 수 있습니다. 재개 시 정책을 다시 검사합니다. v1은 호스트와 모델을 구분하지 않습니다. diff --git a/docs/ko/behavior-differences.md b/docs/ko/behavior-differences.md index 5ea839e..c517b29 100644 --- a/docs/ko/behavior-differences.md +++ b/docs/ko/behavior-differences.md @@ -44,6 +44,6 @@ Runner는 대상 파일을 스냅샷으로 저장하고 도우미의 적용 호 ## 쓰기 점유와 전송 -명령 하나가 실행 중이면 같은 작업 공간의 다른 패치·명령은 `WORKSPACE_BUSY`로 거부됩니다. 대기 큐는 없습니다. 읽기와 검색은 가능합니다. 패치 작업 키는 저장된 요청과 인자가 일치할 때만 이전 결과를 재사용합니다. 응답이 불확실한 상태에서 새 키를 선택하면 변경을 중복 적용할 수 있습니다. +명령 하나가 실행 중이면 같은 작업 공간의 다른 패치·명령은 `WORKSPACE_BUSY`로 거부됩니다. 이미 그 프로세스 뒤에 줄 선 대기자도 마찬가지입니다. 요청이 소유한 패치·exec는 `acquire()`에서 시작하는 인메모리 FIFO에서 현재 요청이 끝날 때까지 기다립니다. 큐 포화는 `RESOURCE_QUEUE_FULL`입니다. 점유 큐는 디스크에 남지 않습니다. 읽기와 검색은 가능합니다. 패치 작업 키는 저장된 요청과 인자가 일치할 때만 이전 결과를 재사용합니다. 응답이 불확실한 상태에서 새 키를 선택하면 변경을 중복 적용할 수 있습니다. stdio와 Streamable HTTP의 도구 스키마는 같습니다. 연결 실패만으로 변경 작업이 실행되었는지 판단할 수 없습니다. [오류 코드](error-codes.md)와 [복구 규칙](agent-integration.md)을 참고하세요. diff --git a/docs/ko/error-codes.md b/docs/ko/error-codes.md index f7d6dff..35de1f4 100644 --- a/docs/ko/error-codes.md +++ b/docs/ko/error-codes.md @@ -25,7 +25,8 @@ | --- | --- | | `UNAUTHORIZED` | 도구 정책이 행동을 거부함 | | `WORKSPACE_NOT_FOUND` | 작업 공간 ID가 등록되지 않음 | -| `WORKSPACE_BUSY` | 다른 변경 작업이나 실행 중인 명령이 작업 공간을 점유함 | +| `WORKSPACE_BUSY` | 라이브 프로세스가 작업 공간 변경 임대를 보유함 | +| `RESOURCE_QUEUE_FULL` | 한 자원의 요청 소유 대기자가 한도에 도달함 | | `INVALID_PATCH` | 패치 파싱·사전 검증·결과 검증 실패. 일부 도우미·입력 오류 경로에도 남아 있음 | | `INVALID_COMMAND` | 잘못된 명령 인자 배열을 실행 전에 거부함 | | `PROCESS_SPAWN_FAILED` | 관리 프로세스를 시작하지 못한 것으로 백엔드가 확인함 | @@ -56,6 +57,7 @@ | `APPROVAL_NOT_FOUND` | 알 수 없는 확인 홀드 ID | | `APPROVAL_CONFLICT` | 홀드가 아직 대기 중이거나 이미 결정되었거나, 재개가 이미 진행 중임 | | `APPROVAL_AMBIGUOUS` | 재개가 중단되어 단말 결과가 디스크에 없음. `approval_id` 포함. 패치는 `operation_id`도 있을 수 있음 | +| `INTERNAL` | 스케줄러·스토어 불변조건 실패. 점유가 아니므로 `WORKSPACE_BUSY`로 재시도하지 말 것 | ## 실행 요청과 완료의 구분 diff --git a/docs/ko/execution-substrate.md b/docs/ko/execution-substrate.md index 705d3e3..00c8c85 100644 --- a/docs/ko/execution-substrate.md +++ b/docs/ko/execution-substrate.md @@ -30,7 +30,7 @@ | 권한 프로필 | 파일·프로세스 행동의 허용 범위를 게이트웨이가 결정 | | 패치 작업 | `operation_id`와 선택적 중복 실행 방지 키, `files`/`changes` 해시, minted/finished 이벤트로 저장하는 패치 원장. `operation_status`로 조회. 명령 실행은 추적하지 않음 | | 프로세스 | 서버가 발급하는 명령 핸들. 메모리에만 보관 | -| 확인 홀드 | 운영자 `approvals` 설정. `approvals` 테이블 행(`pending`/`granted`/`resuming`, 단말 결과가 있을 때까지). 패치 작업이 아니며 권한 부여나 격리 경계도 아님 | +| 확인 홀드 | 운영자 `approvals` 설정. `approvals` 테이블 행(`pending`/`granted`/`queued`/`resuming`, 단말 결과가 있을 때까지). 패치 작업이 아니며 권한 부여나 격리 경계도 아님 | | 논리적 작업 | 작업과 사용자 지시 큐. 전송 세션과 별개 | `environment_id`는 MCP 도구 인자가 아닙니다. 모델이 전달한 프록시 URL이나 Codex 사용자 설정이 실행 권한의 근거가 되지 않습니다. @@ -46,7 +46,7 @@ `exec_command`는 디스패치 식별(`process_id`, `dispatch_status`)만 반환합니다. 종료 판정은 `process_status`의 `running`/`exited`와 termination 메타데이터를 사용합니다. 실행 중인 PTY 크기는 `process_resize`로 바꿉니다. `read_process`는 `output_lost`와 `retained_from`을 포함한 출력을 반환합니다. EOF는 성공이 아닙니다. 핸들이 만료된 뒤의 조회는 새 상태가 아니라 `PROCESS_NOT_FOUND`입니다. 파이프 프로세스의 크기 변경은 `PROCESS_NOT_TTY`, 종료된 핸들은 `PROCESS_NOT_RUNNING`입니다. Linux 샌드박스에서 wait 상태는 관리 자식(헬퍼 argv)의 코드이며, 사용자 argv와 동일하다고 문서화하지 않습니다. -작업 공간 잠금은 패치와 명령이 동시에 파일을 변경하지 못하게 합니다. 명령 실행 중에도 읽기와 검색은 가능하므로 파일 I/O는 사전 경로 검사에만 의존하지 않고 파일을 여는 시점의 심볼릭 링크 변경도 거부해야 합니다. Runner 파일 작업은 `codespace-fs`, 패치 적용은 별도 패치 도우미를 사용합니다. `operation_status`는 기록된 패치 원장(`kind`는 `patch`)을 조회하며, 실행 중인 명령은 `process_id`로만 다루고 이 조회로 복구하지 않습니다. +작업 공간 잠금은 패치와 명령이 동시에 파일을 변경하지 못하게 합니다. FIFO 순서는 MCP 메시지 도착이 아니라, 적격 요청이 자원을 `acquire()`할 때 시작됩니다. 같은 작업 공간에서 요청이 소유한 패치·exec는 현재 요청이 임대를 놓을 때까지 메모리 FIFO로 기다립니다. exec가 라이브 프로세스가 되면 이미 줄 서 있던 대기자와 이후 도착은 `WORKSPACE_BUSY`로 거절됩니다. 큐가 가득 차면 `RESOURCE_QUEUE_FULL`입니다. 이 대기는 SQLite나 스레드 키 큐가 아닙니다. 명령 실행 중에도 읽기와 검색은 가능하므로 파일 I/O는 사전 경로 검사에만 의존하지 않고 파일을 여는 시점의 심볼릭 링크 변경도 거부해야 합니다. Runner 파일 작업은 `codespace-fs`, 패치 적용은 별도 패치 도우미를 사용합니다. `operation_status`는 기록된 패치 원장(`kind`는 `patch`)을 조회하며, 실행 중인 명령은 `process_id`로만 다루고 이 조회로 복구하지 않습니다. UDS 전송과 Linux 샌드박스 준비는 서로 다른 프로토콜과 실패 경계를 가집니다. UDS 변경 요청이 일부만 전달되면 결과가 불확실할 수 있습니다. 연결이 끊겼다는 이유만으로 새 변경 요청을 보내지 마세요. 프로세스 수명은 러너 인스턴스가 소유합니다. MCP/HTTP 클라이언트 끊김은 프로세스를 유지하고, UDS 게이트웨이↔worker 단절이나 게이트웨이 종료는 소유 서브트리를 종료합니다. 영속적인 프로세스 복구는 없습니다. 프로세스·격리 규칙 전체는 [러너 격리](runner-isolation.md)에 설명합니다. @@ -54,6 +54,9 @@ UDS 전송과 Linux 샌드박스 준비는 서로 다른 프로토콜과 실패 + +점유는 자원별 인메모리 FIFO입니다. FIFO 순서는 `acquire()`에서 시작합니다. 요청 소유 exclusive 대기자는 현재 요청이 임대를 놓으면 순서대로 실행됩니다. 확정된 라이브 프로세스는 장벽입니다. 이미 줄 선 대기자와 새 acquire는 `WORKSPACE_BUSY`를 받습니다. spawn 예약은 그 장벽이 아니며, spawn 실패 시 다음 대기자를 깨웁니다. 큐 깊이는 제한되며 포화는 `RESOURCE_QUEUE_FULL`입니다. `read`와 `find`는 shared 임대를 잡지 않습니다. 이 큐는 SQLite에 저장되지 않고 스레드 키로도 구분하지 않습니다. 라이브 MCP 변경은 경로 단위 잠금이 아니라 작업 공간 exclusive를 사용합니다. + @@ -74,11 +77,11 @@ Runner는 해당 작업 공간에서 처음 `read`·`find`·`version`·`apply_pa `approval_create`, `approval_resolve`, `operation_resume`을 호출할 수 있습니다. 작업 공간 프로필이 이미 허용한 변경을 홀드가 승인될 때까지 멈춥니다. 보안 경계가 아닙니다. 권한을 높이거나 `{ "network": true }`·`ClientClaims.approved`를 적용하거나 패치 원장에 V4A 스냅샷을 넣지 않습니다. 같은 MCP 호출자가 grant할 수 있습니다. -운영자가 작업 공간 `approvals`를 `confirm`으로 두면, 정책이 허용한 `apply_patch`와 `exec_command`는 `begin()`이나 프로세스 시작 전에 `APPROVAL_REQUIRED`와 `approval_id`를 반환합니다. 같은 논리 요청을 다시 보내면 그 활성 홀드를 재사용합니다. 기본값 `off`에서는 해당 도구가 바로 실행됩니다. 세 도구는 목록에 남아 있으므로 명시적 `approval_create`로 홀드를 만들 수 있습니다. grant는 프로필을 바꾸지 않습니다. 재개는 `granted`를 `resuming`으로 옮긴 뒤 `allow()`를 다시 검사하고 기존 패치·실행 내부 경로를 돌립니다. `consumed`는 단말 결과와 함께만 기록됩니다. 이후 재개는 그 결과, 패치 원장 복구, 또는 `APPROVAL_AMBIGUOUS`를 반환합니다. 중단된 exec는 다시 spawn하지 않습니다. 정책 거절은 그대로 `UNAUTHORIZED`입니다. 명령은 `process_id`로 다룹니다. v1은 호스트와 모델을 구분하지 않습니다. 서버가 보장하는 것은 재개 시 정책 재검사와 이 내구성 계약입니다. +운영자가 작업 공간 `approvals`를 `confirm`으로 두면, 정책이 허용한 `apply_patch`와 `exec_command`는 `begin()`이나 프로세스 시작 전에 `APPROVAL_REQUIRED`와 `approval_id`를 반환합니다. 같은 논리 요청을 다시 보내면 그 활성 홀드를 재사용합니다. 기본값 `off`에서는 해당 도구가 바로 실행됩니다. 세 도구는 목록에 남아 있으므로 명시적 `approval_create`로 홀드를 만들 수 있습니다. grant는 프로필을 바꾸지 않습니다. 재개는 `granted`를 `queued`로 옮긴 뒤 `allow()`를 다시 검사하고, 자원을 받은 다음에야 `resuming`으로 올립니다. `consumed`는 단말 결과와 함께만 기록됩니다. `queued`에서 재시작하면 다시 acquire합니다. `resuming`에서 이후 재개는 그 결과, 패치 원장 복구, 또는 `APPROVAL_AMBIGUOUS`를 반환합니다. 중단된 exec는 다시 spawn하지 않습니다. 정책 거절은 그대로 `UNAUTHORIZED`입니다. 명령은 `process_id`로 다룹니다. v1은 호스트와 모델을 구분하지 않습니다. 서버가 보장하는 것은 재개 시 정책 재검사와 이 내구성 계약입니다. ## 아직 제공하지 않는 기능 -영속적인 프로세스 복구, 컨테이너·원격 실행, 자원 큐 스케줄러도 제공하지 않습니다. MCP `fs/watch`, UDS watch 이벤트, 쓰기 원인 분류도 제공하지 않습니다. 내부 타입이나 협상된 프로토콜 플래그가 존재한다고 해당 기능을 호출할 수 있는 것은 아닙니다. +영속적인 프로세스 복구와 컨테이너·원격 실행은 제공하지 않습니다. MCP `fs/watch`, UDS watch 이벤트, 쓰기 원인 분류도 제공하지 않습니다. 내부 타입이나 협상된 프로토콜 플래그가 존재한다고 해당 기능을 호출할 수 있는 것은 아닙니다. ## 구현 경계 유지 diff --git a/docs/ko/operations.md b/docs/ko/operations.md index 2548ade..a21c8c3 100644 --- a/docs/ko/operations.md +++ b/docs/ko/operations.md @@ -60,7 +60,7 @@ CodeSpace 저장소의 `workspaces.json`으로 저장합니다. `read-only`는 `network` 기본값은 `restricted`입니다. `enabled`를 사용하려면 Linux 도우미가 필요하며, 지원되는 HTTP 통신은 관리 프록시를 거칩니다. 호스트 네트워크에 무제한 접근하는 설정이 아닙니다. 도우미를 사용할 수 없으면 `enabled` 실행은 실패합니다. `restricted`이고 도우미가 없으면 호스트 실행은 가능하지만 네트워크 제한은 OS 수준에서 강제되지 않습니다. 실제 환경의 적합성은 응답의 정책 집행 상태를 확인해 판단하세요. -`approvals` 기본값은 `off`이며, 정책이 허용한 패치와 명령을 바로 실행합니다. `confirm`이면 `begin()`이나 프로세스 시작 전에 해당 도구를 홀드하고 `APPROVAL_REQUIRED`와 `approval_id`를 반환합니다. 같은 패치나 exec를 다시 보내면 활성 홀드를 재사용합니다. `network`와 같은 운영자 JSON이며 MCP 도구 인자가 아니고 프로필을 올리지 않습니다. 확인 행은 패치 원장과 다른 `approvals` 테이블에 저장되며 `CODESPACE_OPERATIONS_DB`를 같이 씁니다. 홀드가 `pending`·`granted`·`resuming`인 동안 이 테이블은 V4A 패치나 exec argv를 보관합니다. `denied` 또는 `consumed` 뒤에는 본문을 digest 메타(도구, 작업 공간, fingerprint)로 바꿉니다. 패치 원장은 패치 본문이 아니라 해시를 보관합니다. 확인과 재개는 [Agent Loop 연동](agent-integration.md)을 참고하세요. +`approvals` 기본값은 `off`이며, 정책이 허용한 패치와 명령을 바로 실행합니다. `confirm`이면 `begin()`이나 프로세스 시작 전에 해당 도구를 홀드하고 `APPROVAL_REQUIRED`와 `approval_id`를 반환합니다. 같은 패치나 exec를 다시 보내면 활성 홀드를 재사용합니다. `network`와 같은 운영자 JSON이며 MCP 도구 인자가 아니고 프로필을 올리지 않습니다. 확인 행은 패치 원장과 다른 `approvals` 테이블에 저장되며 `CODESPACE_OPERATIONS_DB`를 같이 씁니다. 홀드가 `pending`·`granted`·`queued`·`resuming`인 동안 이 테이블은 V4A 패치나 exec argv를 보관합니다. `denied` 또는 `consumed` 뒤에는 본문을 digest 메타(도구, 작업 공간, fingerprint)로 바꿉니다. 패치 원장은 패치 본문이 아니라 해시를 보관합니다. 확인과 재개는 [Agent Loop 연동](agent-integration.md)을 참고하세요. @@ -133,7 +133,7 @@ worker는 같은 호스트에서 실행하는 별도 프로세스이며 컨테 로그와 데이터베이스는 토큰·게이트웨이 설정과 같이 관리 대상 작업 공간 밖에 두세요. stderr 로그의 보관·순환은 운영자가 관리합니다. Bearer 토큰을 로그나 커밋에 넣지 마세요. 데이터베이스를 삭제하면 패치 중복 실행 방지 기록과 확인 홀드 행도 사라집니다. -패치 응답을 받지 못했다면 `operation_id` 또는 `operation_key` 중 하나만 지정해 `operation_status`를 조회합니다. 재시작 후 미완료 기록은 `unknown`이 되므로 파일을 확인한 뒤 다음 행동을 결정하세요. 프로세스는 `process_id`로 관리하며 `operation_status`로 복구할 수 없습니다. 프로세스가 죽을 때 `resuming`이던 확인 홀드는 패치 원장에서 복구하거나, 저장된 단말 결과를 재현하거나, `APPROVAL_AMBIGUOUS`를 반환할 수 있습니다. exec는 다시 spawn하지 않습니다. [재시도와 복구 규칙](agent-integration.md)을 참고하세요. +패치 응답을 받지 못했다면 `operation_id` 또는 `operation_key` 중 하나만 지정해 `operation_status`를 조회합니다. 재시작 후 미완료 기록은 `unknown`이 되므로 파일을 확인한 뒤 다음 행동을 결정하세요. 프로세스는 `process_id`로 관리하며 `operation_status`로 복구할 수 없습니다. 프로세스가 죽을 때 `queued`이던 확인 홀드는 다시 acquire합니다. `resuming`이던 홀드는 패치 원장에서 복구하거나, 저장된 단말 결과를 재현하거나, `APPROVAL_AMBIGUOUS`를 반환할 수 있습니다. exec는 다시 spawn하지 않습니다. [재시도와 복구 규칙](agent-integration.md)을 참고하세요. @@ -147,6 +147,7 @@ worker는 같은 호스트에서 실행하는 별도 프로세스이며 컨테 | `WORKSPACE_NOT_FOUND` | 레지스트리 경로, 등록 ID, 서버에 전달된 환경변수 | | 패치 도우미 시작 실패 | `codespace-patch` 빌드 여부와 절대 경로 | | `WORKSPACE_BUSY` | 기존 명령이 끝났는지 확인하거나 종료한 뒤 패치 | +| `RESOURCE_QUEUE_FULL` | 대기자가 한도 초과. 잠시 뒤 재시도 | | 명령 시간 초과 | 운영자 제한 시간. 긴 빌드가 완료되었다고 가정하지 않기 | | Linux에서 샌드박스가 없다고 표시 | 도우미 위치, bubblewrap, 네임스페이스 지원 확인 | | `enabled` 명령 시작 실패 | 사용 가능한 Linux 샌드박스 도우미 필요 | diff --git a/docs/ko/security-model.md b/docs/ko/security-model.md index 2cb5971..dfdfbb9 100644 --- a/docs/ko/security-model.md +++ b/docs/ko/security-model.md @@ -42,9 +42,9 @@ Linux 도우미가 없으면 restricted 정책의 작업 공간에서도 비격 ## 작업과 복구의 안전성 -버전 검사는 예상하지 못한 파일 버전에 패치를 적용하는 것을 막습니다. 작업 키는 패치 중복 요청을 구분하며 인증 토큰이 아닙니다. 작업 공간 점유는 파일을 변경할 수 있는 패치·명령 실행을 직렬화합니다. 큐 스케줄러와 영속적인 프로세스 복구는 없습니다. +버전 검사는 예상하지 못한 파일 버전에 패치를 적용하는 것을 막습니다. 작업 키는 패치 중복 요청을 구분하며 인증 토큰이 아닙니다. 작업 공간 점유는 파일을 변경할 수 있는 패치·명령 실행을 직렬화합니다. 요청 소유 충돌은 `acquire()`에서 시작하는 인메모리 FIFO에서 기다리고, 라이브 프로세스는 뒤 대기자와 새 요청에 `WORKSPACE_BUSY`를 반환합니다. 점유 상태는 재시작 후에도 남지 않습니다. 영속적인 프로세스 복구는 없습니다. -`CODESPACE_OPERATIONS_DB`의 `approvals` 확인 홀드는 행이 `pending`·`granted`·`resuming`인 동안 V4A 패치나 exec argv를 보관합니다. 패치 원장은 패치 본문이 아니라 해시를 보관합니다. `denied` 또는 `consumed` 뒤에는 홀드 본문을 digest 메타(도구, 작업 공간, fingerprint)로 바꿉니다. 이 데이터베이스는 토큰·게이트웨이 설정과 같이 작업 공간 루트 밖에 두세요. 홀드는 워크플로 일시정지이며 같은 MCP 호출자가 grant할 수 있습니다. 격리 경계가 아닙니다. +`CODESPACE_OPERATIONS_DB`의 `approvals` 확인 홀드는 행이 `pending`·`granted`·`queued`·`resuming`인 동안 V4A 패치나 exec argv를 보관합니다. 패치 원장은 패치 본문이 아니라 해시를 보관합니다. `denied` 또는 `consumed` 뒤에는 홀드 본문을 digest 메타(도구, 작업 공간, fingerprint)로 바꿉니다. 이 데이터베이스는 토큰·게이트웨이 설정과 같이 작업 공간 루트 밖에 두세요. 홀드는 워크플로 일시정지이며 같은 MCP 호출자가 grant할 수 있습니다. 격리 경계가 아닙니다. 패치 스냅샷 복원은 가능한 범위에서 수행하며 모든 실패의 롤백을 보장하지 않습니다. `unknown`, 부분 실패, 적용 후 검증 오류가 발생하면 해당 파일을 확인하세요. [패치 동작](behavior-differences.md)과 [연동 복구 규칙](agent-integration.md)을 참고하세요. diff --git a/docs/operations.md b/docs/operations.md index db45ce3..e740e8c 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -56,7 +56,7 @@ Save this as `workspaces.json` in the CodeSpace checkout. `read-only` permits re `network` defaults to `restricted`. `enabled` requires the Linux helper and routes supported HTTP traffic through its managed proxy; it does not grant unrestricted host networking. Without a working helper, `enabled` execution fails. With `restricted` and no helper, host execution is possible but network restrictions are not OS-enforced. Use the reported enforcement state when deciding whether an environment is suitable. -`approvals` defaults to `off`, which runs policy-allowed patches and commands immediately. `confirm` holds those tools before `begin()` or spawn and returns `APPROVAL_REQUIRED` with an `approval_id`. Retrying the same patch or exec reuses the active hold. It is operator JSON like `network`, not an MCP argument, and it does not escalate the profile. Confirmation rows share `CODESPACE_OPERATIONS_DB` in an `approvals` table, separate from the patch operations ledger. While a hold is `pending`, `granted`, or `resuming`, that table stores the V4A patch or exec argv. After `denied` or `consumed`, the body is replaced with digest metadata (tool, workspace, fingerprint). The patch ledger keeps hashes, not patch text. See [Agent Loop integration](agent-integration.md) for resolve and resume. +`approvals` defaults to `off`, which runs policy-allowed patches and commands immediately. `confirm` holds those tools before `begin()` or spawn and returns `APPROVAL_REQUIRED` with an `approval_id`. Retrying the same patch or exec reuses the active hold. It is operator JSON like `network`, not an MCP argument, and it does not escalate the profile. Confirmation rows share `CODESPACE_OPERATIONS_DB` in an `approvals` table, separate from the patch operations ledger. While a hold is `pending`, `granted`, `queued`, or `resuming`, that table stores the V4A patch or exec argv. After `denied` or `consumed`, the body is replaced with digest metadata (tool, workspace, fingerprint). The patch ledger keeps hashes, not patch text. See [Agent Loop integration](agent-integration.md) for resolve and resume. @@ -125,7 +125,7 @@ The worker runs on the same host and is not a container. The gateway creates a p Store logs and the database outside the managed workspace, with tokens and gateway configuration. Rotate stderr capture yourself. Do not log Bearer tokens or commit real credentials. Deleting the database also deletes patch idempotency records and confirmation-hold rows. -After losing a patch response, query `operation_status` with exactly one of `operation_id` or `operation_key`. An unfinished record becomes `unknown` after restart; inspect files before deciding what to do. Processes use `process_id` and cannot be recovered through `operation_status`. A confirmation hold that was `resuming` when the process died may recover a patch from that ledger, replay a stored terminal result, or return `APPROVAL_AMBIGUOUS`; exec is not respawned. See [retry and recovery rules](agent-integration.md). +After losing a patch response, query `operation_status` with exactly one of `operation_id` or `operation_key`. An unfinished record becomes `unknown` after restart; inspect files before deciding what to do. Processes use `process_id` and cannot be recovered through `operation_status`. A confirmation hold that was `queued` when the process died can be resumed and will reacquire. A hold that was `resuming` may recover a patch from that ledger, replay a stored terminal result, or return `APPROVAL_AMBIGUOUS`; exec is not respawned. See [retry and recovery rules](agent-integration.md). @@ -137,6 +137,7 @@ After losing a patch response, query `operation_status` with exactly one of `ope | `WORKSPACE_NOT_FOUND` | Registry path, registered ID, and process environment | | Patch helper cannot start | Build `codespace-patch` and check its absolute path | | `WORKSPACE_BUSY` | Finish or terminate the existing command before patching | +| `RESOURCE_QUEUE_FULL` | Too many waiters; retry later | | Command times out | Operator timeout; do not assume a long build completed | | Linux reports no sandbox | Helper location, bubblewrap, namespace support; see isolation guide | | `enabled` command fails before starting | A working Linux sandbox helper is required | diff --git a/docs/security-model.md b/docs/security-model.md index ba956d8..b220d4e 100644 --- a/docs/security-model.md +++ b/docs/security-model.md @@ -32,9 +32,9 @@ The operator-registered root is a trust anchor. Keep tokens, the operations data ## Operation and recovery safety -Version checks prevent applying a patch to an unexpected file version. Operation keys detect duplicate patch requests; they are not authorization tokens. Workspace occupancy serializes mutating patch and exec work. There is no queue scheduler or durable process recovery. +Version checks prevent applying a patch to an unexpected file version. Operation keys detect duplicate patch requests; they are not authorization tokens. Workspace occupancy serializes mutating patch and exec work. Request-owned conflicts wait in an in-memory FIFO that starts at acquire. A live process returns `WORKSPACE_BUSY` for trailing waiters and new arrivals. Occupancy is not durable. There is no durable process recovery. -Confirmation holds in `CODESPACE_OPERATIONS_DB` (`approvals`) store the V4A patch or exec argv while the row is `pending`, `granted`, or `resuming`. The patch operations ledger stores hashes, not patch text. After `denied` or `consumed`, the hold body is replaced with digest metadata (tool, workspace, fingerprint). Keep this database outside the workspace root, with tokens and gateway configuration. The hold is a workflow pause: the same MCP caller can grant it. It is not an isolation boundary. +Confirmation holds in `CODESPACE_OPERATIONS_DB` (`approvals`) store the V4A patch or exec argv while the row is `pending`, `granted`, `queued`, or `resuming`. The patch operations ledger stores hashes, not patch text. After `denied` or `consumed`, the hold body is replaced with digest metadata (tool, workspace, fingerprint). Keep this database outside the workspace root, with tokens and gateway configuration. The hold is a workflow pause: the same MCP caller can grant it. It is not an isolation boundary. Patch snapshot restoration is best effort and is not an all-failure rollback guarantee. On `unknown`, partial failure, or post-apply verification error, inspect affected files. See [patch behavior](behavior-differences.md) and [integration recovery rules](agent-integration.md). diff --git a/docs/translations.json b/docs/translations.json index 0c0073d..e10bdf9 100644 --- a/docs/translations.json +++ b/docs/translations.json @@ -100,8 +100,8 @@ "재시도와-복구", "파일-읽기와-패치" ], - "source_sha256": "45bf2a02b2080ae47d33d0c5ce3d9a5881ee7aec765d85a206573f3d9cb9c8a1", - "translation_sha256": "f189f125e0def02665ff01ddc45abd5fef372d44cfddd7bf5345a19c8ec9cbfd" + "source_sha256": "0788376edef3e4fb7675bf5847df96c3cd97070b77a4b4c331065aeaded82e43", + "translation_sha256": "65bee059368ba91115db048be556d38d7f94c945e0d18c64f160f1db15b5091e" }, { "id": "operations", @@ -152,8 +152,8 @@ "작업-공간-등록", "첫-연결-확인" ], - "source_sha256": "db6477f49619dfdb0582a74a549731f1b4073607437913e7210b012c4a45ea20", - "translation_sha256": "ca90d0dc3900b9e24bf1d4679a0c43a8b9c724233d584ce5fc2c7315b10a68ac" + "source_sha256": "d98494a628e5bb6a717f4ccf75836e92235764e25f448dbcc2d75d94695cc690", + "translation_sha256": "387686ca418ee694e316441eb2dc047d85d0d2e304aece6090ff5d5016051a40" }, { "id": "chatgpt-connector", @@ -240,8 +240,8 @@ "현재-실행-구조", "확장-시-유지할-경계" ], - "source_sha256": "e7bc2b2a86075d17f7eecb672f979dc239b5463688db976277303c511be7f19c", - "translation_sha256": "555724dfeaabd4c7cc7d8e2beafa92c15235389bc009047cd2e064d91ca9bec2" + "source_sha256": "8779e0782981b88d6fed485336f416d65cfbd544d6b676fc82d811a440781557", + "translation_sha256": "b1836fb0915aa764f7853988a14c1e23fc5f8a9126fc8306a370d2aa15295eff" }, { "id": "execution-substrate", @@ -303,8 +303,8 @@ "확인-홀드", "훅과-스킬" ], - "source_sha256": "c577ea9c3ac8b682a40b440a152383eb6e930e7436a740c76f7f4d6f961de832", - "translation_sha256": "8e6bb3c55e8a1656f1cd5c05a4ef2715a6ea2d4b9dbac01134e82e857ced33de" + "source_sha256": "857b87bfff3aa462ea53e17cd9400e77504e7560390f81acfd7ee740ce6b76cb", + "translation_sha256": "6a30c1c8e9ce33c0d08a2c020d1e7e6b480a4eabf96a2e060637f1463436a18d" }, { "id": "protocol-compatibility", @@ -368,8 +368,8 @@ "패치-동작과-복구", "패치-요청-계약" ], - "source_sha256": "8fccab8d541578521e114728c7775d9d9753cf948d7477c9f3e4b17cbecf58a7", - "translation_sha256": "d07978bf886453100b35f4856968b03302a2745776e22a4466d83c27656ed220" + "source_sha256": "102813160616b25115509ed7e933c13d2c1fe7dbda9ebc02867b83c5b6e30d66", + "translation_sha256": "4e300f7f9bc71cb62b588d255a55c2bda97335d76835d760aedaaf17ae6e8e61" }, { "id": "security-model", @@ -414,8 +414,8 @@ "프로세스-정직성", "프로필-mvp" ], - "source_sha256": "c96676f26bb76b12fdd7f9a73c6a114f7b8dda40d798a070e6194602f642893d", - "translation_sha256": "11363cc1c6a3b0b94cc4d58e7ece7a57d27db4ecf3d509bce6e378847eaf4861" + "source_sha256": "76c0e770f66e5fdffce7c6ec21be14ea4204e801fc3f373c9f63687be9d3c9b2", + "translation_sha256": "be0b5e577dcf358df4e453f201e915adbeda9465b2d73a3ddad9c92c22cb3ba0" }, { "id": "runner-isolation", @@ -474,8 +474,8 @@ "전송-실패", "전송-실패-작업-없음" ], - "source_sha256": "a366946fea6e73966595001b72221ae5b9cfe84ddfd77cabba928c4283168fce", - "translation_sha256": "6b97e91bfd708101aedde02ed0ac6f965a09a1136062faa6f5fa3d107d4f26bf" + "source_sha256": "04e3006bc1d7eed2b8a94f44b71dbaa417fceb23f168d0ed4b480825af0c8629", + "translation_sha256": "f286fa26fa1055491de9245c537716ada252177a0d699b7dda37ecd20eee485c" }, { "id": "codex-reuse",