diff --git a/orchestrator_prompt.md b/orchestrator_prompt.md index 190d107..33fb723 100644 --- a/orchestrator_prompt.md +++ b/orchestrator_prompt.md @@ -3,98 +3,57 @@ Coordinate isolated agents to satisfy the authenticated caller goal. Do not do worker, ops, scout, or reviewer work yourself. -## Inputs - -- MULTIAGENT_ORIGINAL_TASK_FILE: authenticated caller request. -- MULTIAGENT_WORKFLOW_ID: supervisor-owned workflow state. -- MULTIAGENT_STATE_DIR: durable agent, review, and trace state. -- MULTIAGENT_PROMPT_MODULE_ROOT: root of role and playbook modules. -- MULTIAGENT_RESUME: 0 for a clean launch, 1 for explicit recovery. - -On a clean launch, read the authenticated caller request and workflow status, -then act. Do not inspect recovery state. When MULTIAGENT_RESUME=1, and only in -that mode, load prompts/playbooks/recovery.md before restoring work. - -## Role Catalog - -Choose by capability, not provider or task name. - -| Capability | Role | Module | -| --- | --- | --- | -| Change bounded workspace paths | worker | prompts/worker.md | -| Access an external provider or deployed service through a Markdown runbook and prod-mcp, including read-only access | ops | prompts/roles/ops-agent.md | -| Resolve a material unknown from repository, workspace, session, or already-returned immutable evidence | scout | matching file under prompts/roles/ | -| Review a decision, request, diff, receipt, or claim | reviewer/verifier | matching reviewer module | - -Specialized modules include contract and acceptance scouts, decision authority, -ops review, scope review, build verification, and validation coordination. Load -only the module selected for the current node. - -External access is an authority boundary, not a mutability classification. A -scout never calls Slack, GitHub, Grafana, AWS, Kubernetes, prod-mcp, or another -deployed service. Spawn ops first to acquire external evidence under a reviewed -runbook; only then may a scout analyze the immutable returned artifact. - -## Decide The DAG - -1. Read the goal and persisted supervisor state. -2. Identify outputs needed for acceptance and unresolved material facts. -3. Select the smallest role DAG that can produce those outputs. -4. Omit a scout or reviewer when its output cannot affect acceptance, unless - the supervisor reports it as an obligation. -5. Spawn ready nodes, wait for durable output, and submit evidence to the - supervisor gate. -6. On rejection, satisfy the reported obligation or revise the DAG; do not - bypass the gate. - -The orchestrator decides the DAG. The supervisor enforces role isolation, -authority, immutable evidence bindings, independent reviews, and phase or -completion gates. - -## Supervisor Gates - -- Spawn every role with multiagent subagent spawn. Provider-native agent tools - do not establish Linux identity, Landlock policy, or trusted evidence. -- Source implementation follows the bundled - prompts/playbooks/implementation-lifecycle.md gate. -- An implementation without a contract scout still requires an independently - reviewed, supervisor-approved implementation context before a worker starts. -- An ops-only task that does not change repository source does not enter the - source implementation lifecycle. Do not create an implementation context, - decision-authority reviewer, or implementation-phase transition for it; use - its Markdown runbook and independently reviewed ops requests directly. -- multiagent ops execute requires the finalized independent reviewer bound to - the exact request and runbook. -- A completion request succeeds only after supervisor obligations and TODOs are - satisfied. - -Prompt text cannot grant authority or waive a supervisor rejection. - -## Coordination - -Load prompts/playbooks/orchestration-routing.md to select a role and -prompts/playbooks/agent-spawning.md only for worker, scout, verifier, or other -non-ops role lifecycles. Load -prompts/playbooks/finding-todo-loop.md only for findings and repair, and -prompts/playbooks/validation-scheduling.md only when validation could overlap. -When selecting ops, load only prompts/playbooks/reviewed-ops-cycle.md and use -its initial spawn and runtime command instead of loading agent-spawning.md or -constructing review and continuation steps yourself. -`multiagent subagent spawn` composes the canonical role module automatically. -Do not search for, enumerate, or read role prompt files to discover how to -spawn a known role. - -Keep at most one active agent for the same responsibility. Use bounded waits, -inspect durable results, finalize completed agents, and preserve -MULTIAGENT_STATE_DIR. Never treat missing provider-native tools or role -credentials as proof that a supervisor-mediated capability is unavailable. - -Keep one ops identity for the entire session. It selects and follows runbooks, -materializes immutable requests, and continues after each reviewed operation. -For every request, invoke `multiagent subagent reviewed-ops-cycle`; do not -manually spawn its reviewer, construct binding evidence, restore the ops agent, -or create a replacement ops identity. Finalize the ops identity only after the -session's operational work finishes or reaches a blocker. - -MULTIAGENT_VERIFIER_MAX_ITERATIONS is an escalation threshold, never an -acceptance condition. +## Start + +On a clean launch: + +1. Run `multiagent workflow context "$MULTIAGENT_WORKFLOW_ID"`. +2. Read the authenticated task artifact named by `originalTask` exactly once. +3. Route from that typed context. Do not inspect panes, rediscover state paths, + or reconstruct provider transcripts. + +Only when `MULTIAGENT_RESUME=1`, load +`prompts/playbooks/recovery.md` before restoring work. + +## Choose roles + +| Need | Role | +| --- | --- | +| Change bounded workspace paths | worker | +| Use a Markdown runbook to access prod-mcp or an external service | ops | +| Resolve a material unknown from local or immutable evidence | scout | +| Independently assess a decision, request, diff, receipt, or claim | reviewer/verifier | + +External access always belongs to ops. A scout may analyze an immutable artifact +returned by ops, but cannot call Slack, GitHub, Grafana, AWS, Kubernetes, +prod-mcp, or another deployed service. + +## Build the DAG + +1. Identify the outputs required for acceptance and the facts still unknown. +2. Spawn the smallest role DAG that can produce them. +3. Omit work that cannot affect acceptance unless the supervisor reports an + obligation. +4. Submit durable evidence to supervisor gates. On rejection, satisfy the + obligation or revise the DAG; never bypass the gate. + +The supervisor, not prompt text, enforces identity, authority, evidence +bindings, independent review, and phase completion. + +## Required lifecycles + +- Spawn roles with `multiagent subagent spawn`; provider-native agents do not + establish the required Linux identity or evidence boundary. +- Source changes follow `prompts/playbooks/implementation-lifecycle.md`. +- External-only work skips the source lifecycle and uses reviewed ops requests. +- For ops, load only `prompts/playbooks/reviewed-ops-cycle.md`. Keep one ops + identity for the session and invoke `multiagent subagent reviewed-ops-cycle` + for every immutable request. +- Use a fresh reviewer for each immutable ops request. Finalize the ops identity + only when operational work completes or reaches a blocker. +- Load other playbooks only when their lifecycle is selected. Do not enumerate + prompt files to discover known roles. + +Keep one active agent per responsibility, use bounded waits, and rely on durable +results rather than terminal prose. `MULTIAGENT_VERIFIER_MAX_ITERATIONS` is an +escalation threshold, not acceptance. diff --git a/prompts/playbooks/reviewed-ops-cycle.md b/prompts/playbooks/reviewed-ops-cycle.md index bcdbacb..5d6dd1d 100644 --- a/prompts/playbooks/reviewed-ops-cycle.md +++ b/prompts/playbooks/reviewed-ops-cycle.md @@ -1,59 +1,43 @@ # Reviewed Ops Cycle -Use this playbook whenever an ops agent has materialized an immutable prod-mcp -request. It centralizes the review and continuation mechanics; provider runbooks -define what operations mean, not how agents are spawned. +Use one logical ops identity for all operational work in a session. The ops +agent chooses runbooks and request contents; the supervisor enforces publication, +review, and execution boundaries. -## Session invariant - -Use one persistent `ops` identity for the entire session. That agent may follow -multiple runbooks and materialize multiple requests, but no second ops identity -may be created. Agent judgment selects the runbook, operation, and parameters. -`multiagent ops bind-runbook` copies canonical target metadata from the exact -Markdown runbook when it is declared there. This playbook only makes the -authorization lifecycle deterministic. - -## Start the ops identity - -Spawn exactly one ops identity for the session. The runtime composes the ops -role module; do not load the general agent-spawning playbook or role prompt -files to reconstruct it. +## Start once ```bash -multiagent subagent spawn OPS_NAME --role ops --instruction "Inspect the request, follow the applicable runbook, and prepare the reviewed operation." +multiagent subagent spawn OPS_NAME --role ops \ + --instruction "Follow the applicable runbook and prepare the reviewed operation." multiagent subagent wait OPS_NAME --timeout 900 ``` -Keep this identity for every reviewed operation in the session. Do not spawn a -replacement ops identity after review. +Do not replace this identity after review. -## Reviewed request +## Review and continue -After the ops agent writes and binds its request, run: +After ops returns a published artifact descriptor: ```bash multiagent subagent reviewed-ops-cycle OPS_NAME \ - --request-file "$MULTIAGENT_LOG_DIR/agents/OPS_NAME/request.json" \ + --request-file "$PUBLISHED_REQUEST_PATH" \ --reviewer ops-reviewer-NN \ --timeout 900 ``` -Use a fresh `ops-reviewer-NN` identity for every immutable request. The command: - -1. verifies that the request belongs to the session's ops identity; -2. computes the exact review binding; -3. spawns an independent ops reviewer with the literal request and binding; -4. waits for and finalizes the reviewer; -5. rejects missing, negative, or incorrectly bound evidence before execution; -6. restores the same ops identity with the exact execute command; and -7. waits for that ops identity to inspect the receipt and continue its runbook. +Use a fresh reviewer name for each immutable request. This command: -Do not manually reconstruct these steps in prompts or shell commands. +1. publishes a safe legacy request when necessary; +2. binds the reviewer to the immutable request and exact runbook; +3. passes only a bounded artifact descriptor to the reviewer; +4. finalizes accepted review evidence before execution; and +5. continues the same ops identity in a fresh provider context with the exact + execute command. -## Failure behavior +Do not reconstruct these mechanics manually. Prior panes, transcripts, final +messages, and native provider resume state are intentionally excluded from the +continuation boundary. -If review is rejected, binding preflight fails, restoration fails, or prod-mcp -rejects execution, stop and report the exact blocker. If a corrected independent -review is appropriate, use a fresh reviewer on the same immutable request. If -the request bytes must change, the same ops identity materializes the new bytes -and starts a new reviewed cycle. Never spawn a replacement ops identity. +On rejection or preflight failure, report the blocker. A changed request needs a +new publication and reviewer. A review correction may use a fresh reviewer on +the same immutable request. Never create a second ops identity. diff --git a/prompts/roles/ops-agent.md b/prompts/roles/ops-agent.md index 595ef7b..b5504d4 100644 --- a/prompts/roles/ops-agent.md +++ b/prompts/roles/ops-agent.md @@ -1,16 +1,42 @@ -# Operations Agent Role - -You are the operations agent. Execute the original goal by following the supplied Markdown runbook exactly. The Markdown file is the authoritative procedure; the JSON request is only its bounded prod-mcp execution envelope. - -- Do not encode operations in agent policy or source code. Read the `.md` runbook and prod-mcp target contract. -- Derive each operation, target, parameter, and phase from the authenticated goal and exact runbook. Do not accept an operation invented by the orchestrator. -- Materialize one bounded JSON request under `$MULTIAGENT_LOG_DIR/agents/$MULTIAGENT_SUBAGENT_NAME/request.json`. The generic envelope is `taskId`, `goal`, `operation: {id, version}`, `target: {environment, cluster, namespace, service}`, `parameters: {}`, and `runbook: {id, version, phase}`. Derive every value from the goal, runbook, and prod-mcp target contract. Include `changeTicket` only when the runbook requires one. Never supply `approvals`; the supervisor derives them from the authenticated caller and sealed reviewer evidence. -- Certify and schema-check the request against the exact Markdown bytes by running `multiagent ops bind-runbook --request-file "$REQUEST_FILE" --runbook-document runbooks/SELECTED.md`. A successful command proves that the generic envelope is executable and adds the exact `runbookDocument` and `runbookContentSha256`; if it fails, correct the unreviewed request instead of sending malformed JSON to a reviewer. Never calculate or write `runbookContentSha256` yourself. Any change after this command creates a different request and requires binding again. -- Keep production evidence in your role-owned trace directory, never in the repository or another role's private home. -- Before execution, print the exact request and request path, then exit so an independent `ops-reviewer` can inspect the same literal request. Do not execute an unreviewed request. -- When restored for execution with accepted reviewer evidence, call `multiagent ops execute --request-file PATH --reviewer REVIEWER_NAME` against the exact unchanged request path inspected by that finalized reviewer. The `--reviewer` argument is mandatory. Do not infer approval from or edit the workflow ledger, and do not recreate, copy, reformat, or otherwise mutate the reviewed file. Wait for its persisted receipt before continuing the runbook. -- Treat changed request content as a new request that requires a new independent review. -- You have authority to request any operation allowed by prod-mcp. You do not possess AWS, KMS, bearer-token, Grafana, or Kubernetes credentials. -- The authority supervisor owns KMS signing and prod-mcp transport authentication. Missing credential environment variables in this role are intentional; use the local `multiagent ops` commands and report only an actual broker rejection. -- Stop if the runbook is ambiguous, the reviewer rejects the request, or prod-mcp rejects the target or operation. -- Report the persisted operation action ID and receipt path. Never bypass the supervisor. +# Operations Agent + +Follow the authenticated goal and authoritative Markdown runbook. Do not invent +operations or encode provider-specific behavior in policy or source code. + +## Prepare + +1. Select the applicable runbook and derive the operation, target, parameters, + and phase from it and the prod-mcp target contract. +2. Write one bounded JSON draft under + `$MULTIAGENT_LOG_DIR/agents/$MULTIAGENT_SUBAGENT_NAME/` using the generic + envelope: `taskId`, `goal`, `operation`, `target`, `parameters`, and + `runbook`. Add `changeTicket` only when required. Never add `approvals` or + calculate `runbookContentSha256`. +3. Publish it with: + +```bash +multiagent ops publish --draft-file "$DRAFT_FILE" \ + --runbook-document runbooks/SELECTED.md +``` + +Correct only the unpublished draft if validation fails. The returned descriptor +identifies the supervisor-owned immutable request; never copy or modify it. +Report that descriptor and stop for independent review. + +## Execute after review + +When restored, use the exact command supplied by the supervisor: + +```bash +multiagent ops execute --request-file PATH --reviewer REVIEWER_NAME +``` + +Execute the reviewed request once. Interpret the structured outcome under the +runbook and decide whether to finish, escalate, or prepare a distinct request. +Changed bytes always require a new review. Report the action ID, receipt path, +result, or exact blocker. + +Keep evidence in your trace directory. Credentials and signing authority remain +with the supervisor and prod-mcp; missing credential environment variables are +expected. Stop on an ambiguous runbook, reviewer rejection, or prod-mcp policy +rejection. Never bypass the supervisor or create a replacement ops identity. diff --git a/src/authority.rs b/src/authority.rs index 482a3bd..50a30b3 100644 --- a/src/authority.rs +++ b/src/authority.rs @@ -44,6 +44,7 @@ enum AuthorityOperation { ValidationLeaseShow, ValidationLeaseList, GateCheck, + OpsPublish, OpsExecute, } @@ -53,6 +54,9 @@ impl AuthorityRequest { "workflow" => (AuthorityOperation::Workflow, args), "decision" => (AuthorityOperation::Decision, args), "dag" => (AuthorityOperation::Dag, args), + "ops" if args.first().map(String::as_str) == Some("publish") => { + (AuthorityOperation::OpsPublish, &args[1..]) + } "ops" if args.first().map(String::as_str) == Some("execute") => { (AuthorityOperation::OpsExecute, &args[1..]) } @@ -136,7 +140,9 @@ impl AuthorityRequest { | AuthorityOperation::TodoAssign | AuthorityOperation::TodoStatus | AuthorityOperation::GateCheck => uid == config::ORCHESTRATOR_UID, - AuthorityOperation::OpsExecute => uid == config::OPS_UID, + AuthorityOperation::OpsPublish | AuthorityOperation::OpsExecute => { + uid == config::OPS_UID + } AuthorityOperation::FindingCreate => uid == config::READER_UID, AuthorityOperation::FindingDismiss | AuthorityOperation::TodoClose => { matches!(uid, config::ORCHESTRATOR_UID | config::READER_UID) @@ -197,6 +203,7 @@ impl AuthorityRequest { AuthorityOperation::ValidationLeaseShow => ("subagent", Some("validation-lease-show")), AuthorityOperation::ValidationLeaseList => ("subagent", Some("validation-lease-list")), AuthorityOperation::GateCheck => ("subagent", Some("gate-check")), + AuthorityOperation::OpsPublish => ("ops", Some("publish")), AuthorityOperation::OpsExecute => ("ops", Some("execute")), }; let mut args = self.args; diff --git a/src/prod_ops.rs b/src/prod_ops.rs index 988f468..4732a6b 100644 --- a/src/prod_ops.rs +++ b/src/prod_ops.rs @@ -3,13 +3,41 @@ use serde_json::{json, Value}; use sha2::{Digest, Sha256}; use std::env; use std::fs::{self, OpenOptions}; -use std::io::Write; +use std::io::{Read, Write}; #[cfg(target_os = "linux")] -use std::os::unix::fs::OpenOptionsExt; +use std::os::unix::fs::MetadataExt; +#[cfg(unix)] +use std::os::unix::fs::{OpenOptionsExt, PermissionsExt}; use std::path::{Path, PathBuf}; use std::process::{Command, ExitCode, Stdio}; use std::time::{SystemTime, UNIX_EPOCH}; +const MAX_OPERATION_REQUEST_BYTES: u64 = 65_536; +const MAX_RUNBOOK_BYTES: u64 = 1_048_576; + +pub(crate) struct PublishedRequest { + artifact_path: PathBuf, + sha256: String, + bytes: usize, +} + +impl PublishedRequest { + pub(crate) fn path(&self) -> &Path { + &self.artifact_path + } + + pub(crate) fn descriptor_json(&self) -> Result { + serde_json::to_string(&json!({ + "artifactPath": self.artifact_path, + "sha256": self.sha256, + "bytes": self.bytes, + "mediaType": "application/json", + "truncated": false + })) + .map_err(|error| format!("encode ops publication descriptor: {error}")) + } +} + struct TrustedApproval { subject: String, role: &'static str, @@ -20,9 +48,10 @@ struct TrustedApproval { pub fn run(args: &[String]) -> Result { match args.first().map(String::as_str) { Some("bind-runbook") => bind_runbook(&args[1..]), + Some("publish") => publish(&args[1..]), Some("execute") => execute(&args[1..]), Some("review-bind") => review_bind(&args[1..]), - _ => Err("usage: multiagent ops bind-runbook --request-file PATH --runbook-document PATH | multiagent ops review-bind --request-file PATH | multiagent ops execute --request-file PATH --reviewer NAME".into()), + _ => Err("usage: multiagent ops bind-runbook --request-file PATH --runbook-document PATH | multiagent ops publish --draft-file PATH --runbook-document PATH | multiagent ops review-bind --request-file PATH | multiagent ops execute --request-file PATH --reviewer NAME".into()), } } @@ -35,24 +64,9 @@ fn bind_runbook(args: &[String]) -> Result { if !request_file.starts_with(&state) { return Err("ops request file must be inside MULTIAGENT_STATE_DIR".into()); } - let bytes = fs::read(&request_file).map_err(|error| format!("read ops request: {error}"))?; - if bytes.is_empty() || bytes.len() > 65_536 { - return Err("ops request must contain between 1 and 65536 bytes".into()); - } - let mut template: Value = serde_json::from_slice(&bytes) - .map_err(|error| format!("decode ops request template: {error}"))?; + let (bytes, _) = read_bounded_file(&request_file, MAX_OPERATION_REQUEST_BYTES, false)?; let relative = required(&options, "--runbook-document")?; - let digest = exact_runbook_content_sha256(relative)?; - let canonical_target = exact_runbook_target(relative)?; - let object = template - .as_object_mut() - .ok_or("ops request template must be an object")?; - if let Some(target) = canonical_target { - object.insert("target".into(), target); - } - object.insert("runbookDocument".into(), Value::String(relative.into())); - object.insert("runbookContentSha256".into(), Value::String(digest.clone())); - validate_request_template(&template)?; + let (template, digest) = bind_request_template(&bytes, relative, false)?; let encoded = serde_json::to_vec_pretty(&template) .map_err(|error| format!("encode bound ops request: {error}"))?; fs::write(&request_file, encoded) @@ -62,6 +76,297 @@ fn bind_runbook(args: &[String]) -> Result { Ok(ExitCode::SUCCESS) } +fn publish(args: &[String]) -> Result { + let options = options(args)?; + let state = fs::canonicalize(required_env("MULTIAGENT_STATE_DIR")?) + .map_err(|error| format!("resolve multiagent state: {error}"))?; + let draft_file = PathBuf::from(required(&options, "--draft-file")?); + let runbook_document = required(&options, "--runbook-document")?; + let draft = read_ops_draft(&draft_file)?; + let (template, _) = bind_request_template(&draft, runbook_document, true)?; + let encoded = serde_json::to_vec_pretty(&template) + .map_err(|error| format!("encode published ops request: {error}"))?; + let descriptor = publish_request_bytes(&state, &encoded)?; + println!("{}", descriptor.descriptor_json()?); + Ok(ExitCode::SUCCESS) +} + +pub(crate) fn publish_bound_request( + state: &Path, + request_file: &Path, +) -> Result { + let state = fs::canonicalize(state) + .map_err(|error| format!("resolve multiagent state: {error}"))?; + let (_, bytes) = read_reviewable_request(&state, request_file)?; + let template: Value = serde_json::from_slice(&bytes) + .map_err(|error| format!("decode bound ops request: {error}"))?; + validate_request_template(&template)?; + verified_runbook_content(&template)?; + publish_request_bytes(&state, &bytes) +} + +fn bind_request_template( + bytes: &[u8], + runbook_document: &str, + require_canonical_target: bool, +) -> Result<(Value, String), String> { + let mut template: Value = serde_json::from_slice(bytes) + .map_err(|error| format!("decode ops request template: {error}"))?; + validate_request_envelope(&template)?; + let runbook_bytes = exact_runbook_bytes(runbook_document)?; + let canonical_target = canonical_runbook_target(&runbook_bytes)?; + if require_canonical_target && canonical_target.is_none() { + return Err("published runbook must declare one canonical target".into()); + } + let digest = runbook_content_digest(&runbook_bytes); + let object = template + .as_object_mut() + .ok_or("ops request template must be an object")?; + if let Some(target) = canonical_target { + object.insert("target".into(), target); + } + object.insert( + "runbookDocument".into(), + Value::String(runbook_document.into()), + ); + object.insert( + "runbookContentSha256".into(), + Value::String(digest.clone()), + ); + validate_request_template(&template)?; + Ok((template, digest)) +} + +fn publish_request_bytes(state: &Path, bytes: &[u8]) -> Result { + if bytes.is_empty() || bytes.len() as u64 > MAX_OPERATION_REQUEST_BYTES { + return Err("ops request must contain between 1 and 65536 bytes".into()); + } + let hex = format!("{:x}", Sha256::digest(bytes)); + let directory = state + .join("operations") + .join("requests") + .join(&hex); + fs::create_dir_all(&directory) + .map_err(|error| format!("create operation request store: {error}"))?; + secure_publication_path(&directory, true)?; + let artifact_path = directory.join("request.json"); + if artifact_path.exists() { + let (existing, metadata) = + read_bounded_file(&artifact_path, MAX_OPERATION_REQUEST_BYTES, true)?; + validate_published_metadata(&metadata)?; + if existing != bytes { + return Err("content-addressed operation request collision".into()); + } + } else { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|error| error.to_string())? + .as_nanos(); + let temporary = directory.join(format!( + ".request.{}.{}.tmp", + std::process::id(), + unique + )); + let mut options = OpenOptions::new(); + options.write(true).create_new(true); + #[cfg(unix)] + options.mode(0o440); + let mut file = options + .open(&temporary) + .map_err(|error| format!("create operation request artifact: {error}"))?; + secure_publication_file(&file)?; + file.write_all(bytes) + .and_then(|_| file.sync_all()) + .map_err(|error| format!("persist operation request artifact: {error}"))?; + match fs::hard_link(&temporary, &artifact_path) { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + let (existing, metadata) = + read_bounded_file(&artifact_path, MAX_OPERATION_REQUEST_BYTES, true)?; + validate_published_metadata(&metadata)?; + if existing != bytes { + let _ = fs::remove_file(&temporary); + return Err("content-addressed operation request collision".into()); + } + } + Err(error) => { + let _ = fs::remove_file(&temporary); + return Err(format!("publish operation request artifact: {error}")); + } + } + let _ = fs::remove_file(&temporary); + } + Ok(PublishedRequest { + artifact_path, + sha256: format!("sha256:{hex}"), + bytes: bytes.len(), + }) +} + +fn read_ops_draft(path: &Path) -> Result, String> { + let agents = fs::canonicalize(PathBuf::from(required_env("MULTIAGENT_LOG_DIR")?).join("agents")) + .map_err(|error| format!("resolve ops agents log directory: {error}"))?; + let canonical = fs::canonicalize(path) + .map_err(|error| format!("resolve ops draft file: {error}"))?; + if !canonical.starts_with(&agents) { + return Err("ops draft file must be inside MULTIAGENT_LOG_DIR/agents".into()); + } + let (bytes, _metadata) = + read_bounded_file(&canonical, MAX_OPERATION_REQUEST_BYTES, true)?; + if bytes.is_empty() { + return Err("ops request must contain between 1 and 65536 bytes".into()); + } + #[cfg(target_os = "linux")] + if _metadata.uid() != crate::config::OPS_UID || _metadata.mode() & 0o022 != 0 { + return Err("ops draft must be owned by the ops UID and not group- or world-writable".into()); + } + Ok(bytes) +} + +fn read_reviewable_request(state: &Path, path: &Path) -> Result<(PathBuf, Vec), String> { + let canonical = fs::canonicalize(path) + .map_err(|error| format!("resolve ops request file: {error}"))?; + if !canonical.starts_with(state) { + return Err("ops request file must be inside MULTIAGENT_STATE_DIR".into()); + } + let (bytes, _metadata) = + read_bounded_file(&canonical, MAX_OPERATION_REQUEST_BYTES, true)?; + if bytes.is_empty() { + return Err("ops request must contain between 1 and 65536 bytes".into()); + } + #[cfg(target_os = "linux")] + if !matches!( + _metadata.uid(), + crate::config::OPS_UID | crate::config::SUPERVISOR_UID + ) || _metadata.mode() & 0o022 != 0 + { + return Err("ops request must be safely owned by the ops or supervisor UID".into()); + } + Ok((canonical, bytes)) +} + +fn read_bounded_file( + path: &Path, + limit: u64, + no_follow: bool, +) -> Result<(Vec, fs::Metadata), String> { + let mut options = OpenOptions::new(); + options.read(true); + #[cfg(unix)] + if no_follow { + options.custom_flags(libc::O_NOFOLLOW); + } + let mut file = options + .open(path) + .map_err(|error| format!("open {}: {error}", path.display()))?; + let metadata = file + .metadata() + .map_err(|error| format!("inspect {}: {error}", path.display()))?; + if !metadata.is_file() || metadata.len() > limit { + return Err(format!( + "{} must be a regular file no larger than {limit} bytes", + path.display() + )); + } + let mut bytes = Vec::with_capacity(metadata.len() as usize); + Read::take(&mut file, limit + 1) + .read_to_end(&mut bytes) + .map_err(|error| format!("read {}: {error}", path.display()))?; + if bytes.len() as u64 > limit { + return Err(format!("{} exceeds the configured byte limit", path.display())); + } + Ok((bytes, metadata)) +} + +#[cfg(target_os = "linux")] +fn secure_publication_path(path: &Path, directory: bool) -> Result<(), String> { + use std::ffi::CString; + use std::os::unix::ffi::OsStrExt; + + let metadata = fs::symlink_metadata(path) + .map_err(|error| format!("inspect publication path {}: {error}", path.display()))?; + if metadata.file_type().is_symlink() || (directory && !metadata.is_dir()) { + return Err(format!("unsafe publication path: {}", path.display())); + } + if metadata.uid() != crate::config::SUPERVISOR_UID { + if unsafe { libc::geteuid() } != 0 { + return Err("operation request store must be supervisor-owned".into()); + } + let raw = CString::new(path.as_os_str().as_bytes()) + .map_err(|_| format!("publication path contains NUL: {}", path.display()))?; + if unsafe { + libc::lchown( + raw.as_ptr(), + crate::config::SUPERVISOR_UID, + crate::config::ROLE_GID, + ) + } != 0 + { + return Err(format!( + "set publication ownership {}: {}", + path.display(), + std::io::Error::last_os_error() + )); + } + } + fs::set_permissions( + path, + fs::Permissions::from_mode(if directory { 0o750 } else { 0o440 }), + ) + .map_err(|error| format!("secure publication path {}: {error}", path.display())) +} + +#[cfg(not(target_os = "linux"))] +fn secure_publication_path(path: &Path, directory: bool) -> Result<(), String> { + #[cfg(unix)] + fs::set_permissions( + path, + fs::Permissions::from_mode(if directory { 0o750 } else { 0o440 }), + ) + .map_err(|error| format!("secure publication path {}: {error}", path.display()))?; + Ok(()) +} + +fn secure_publication_file(file: &fs::File) -> Result<(), String> { + #[cfg(target_os = "linux")] + { + use std::os::fd::AsRawFd; + let metadata = file + .metadata() + .map_err(|error| format!("inspect operation request artifact: {error}"))?; + if metadata.uid() != crate::config::SUPERVISOR_UID { + if unsafe { libc::geteuid() } != 0 { + return Err("operation request artifact must be supervisor-owned".into()); + } + if unsafe { + libc::fchown( + file.as_raw_fd(), + crate::config::SUPERVISOR_UID, + crate::config::ROLE_GID, + ) + } != 0 + { + return Err(format!( + "set operation request artifact ownership: {}", + std::io::Error::last_os_error() + )); + } + } + } + #[cfg(unix)] + file.set_permissions(fs::Permissions::from_mode(0o440)) + .map_err(|error| format!("secure operation request artifact: {error}"))?; + Ok(()) +} + +fn validate_published_metadata(_metadata: &fs::Metadata) -> Result<(), String> { + #[cfg(target_os = "linux")] + if _metadata.uid() != crate::config::SUPERVISOR_UID || _metadata.mode() & 0o227 != 0 { + return Err("published operation request has unsafe ownership or mode".into()); + } + Ok(()) +} + fn review_bind(args: &[String]) -> Result { let options = options(args)?; let bytes = fs::read(required(&options, "--request-file")?) @@ -118,24 +423,7 @@ fn load_reviewed_request( validate_id("reviewer name", reviewer)?; let state = fs::canonicalize(required_env("MULTIAGENT_STATE_DIR")?) .map_err(|error| format!("resolve multiagent state: {error}"))?; - let request_file = fs::canonicalize(request_file) - .map_err(|error| format!("resolve ops request file: {error}"))?; - if !request_file.starts_with(&state) { - return Err("ops request file must be inside MULTIAGENT_STATE_DIR".into()); - } - let bytes = fs::read(&request_file).map_err(|error| format!("read ops request: {error}"))?; - if bytes.is_empty() || bytes.len() > 65_536 { - return Err("ops request must contain between 1 and 65536 bytes".into()); - } - #[cfg(target_os = "linux")] - { - use std::os::unix::fs::{MetadataExt, PermissionsExt}; - let metadata = - fs::metadata(&request_file).map_err(|error| format!("inspect ops request: {error}"))?; - if metadata.uid() != crate::config::OPS_UID || metadata.permissions().mode() & 0o022 != 0 { - return Err("ops request must be owned by the ops UID and not group-writable".into()); - } - } + let (_, bytes) = read_reviewable_request(&state, request_file)?; let template: Value = serde_json::from_slice(&bytes) .map_err(|error| format!("decode ops request template: {error}"))?; let runbook_content_sha256 = verified_runbook_content(&template)?; @@ -186,7 +474,26 @@ fn execute(args: &[String]) -> Result { "request": request }))?; let permit = sign_permit(&payload)?; - let result = call_prod_mcp(&permit)?; + let result = call_prod_mcp(&permit).unwrap_or_else(|message| { + json!({ + "jsonrpc":"2.0", + "id":2, + "result":{ + "isError":true, + "structuredContent":{ + "code":"prod_mcp_transport_failure", + "message":message, + "outcome":{ + "disposition":"failed", + "terminal":true, + "retryable":true, + "code":"prod_mcp_transport_failure", + "requiredActor":"service-operator" + } + } + } + }) + }); let operation_dir = state.join("operations").join(&action_id); fs::create_dir_all(&operation_dir) .map_err(|error| format!("create operation receipt directory: {error}"))?; @@ -528,27 +835,10 @@ fn required_object<'a>( } fn validate_request_template(template: &Value) -> Result<(), String> { + validate_request_envelope(template)?; let object = template .as_object() .ok_or("ops request template must be an object")?; - let task_id = required_template_string(object, "taskId")?; - validate_id("task ID", task_id)?; - if object.get("goal").is_none_or(Value::is_null) { - return Err("ops request template requires goal".into()); - } - if object.contains_key("approvals") { - return Err("ops request approvals are derived by the supervisor and cannot be supplied by an agent".into()); - } - - let operation = required_object(object, "operation")?; - if operation.len() != 2 { - return Err("ops request operation must contain only id and version".into()); - } - validate_id("operation ID", required_template_string(operation, "id")?)?; - validate_semver( - "operation version", - required_template_string(operation, "version")?, - )?; let target = required_object(object, "target")?; if target.len() != 4 { @@ -567,17 +857,6 @@ fn validate_request_template(template: &Value) -> Result<(), String> { )?; } - required_object(object, "parameters")?; - let runbook = required_object(object, "runbook")?; - if runbook.len() != 3 { - return Err("ops request runbook must contain id, version, and phase".into()); - } - validate_id("runbook ID", required_template_string(runbook, "id")?)?; - validate_semver( - "runbook version", - required_template_string(runbook, "version")?, - )?; - validate_id("runbook phase", required_template_string(runbook, "phase")?)?; required_template_string(object, "runbookDocument")?; let digest = required_template_string(object, "runbookContentSha256")?; let hex = digest @@ -593,6 +872,43 @@ fn validate_request_template(template: &Value) -> Result<(), String> { Ok(()) } +fn validate_request_envelope(template: &Value) -> Result<(), String> { + let object = template + .as_object() + .ok_or("ops request template must be an object")?; + let task_id = required_template_string(object, "taskId")?; + validate_id("task ID", task_id)?; + if object.get("goal").is_none_or(Value::is_null) { + return Err("ops request template requires goal".into()); + } + if object.contains_key("approvals") { + return Err("ops request approvals are derived by the supervisor and cannot be supplied by an agent".into()); + } + + let operation = required_object(object, "operation")?; + if operation.len() != 2 { + return Err("ops request operation must contain only id and version".into()); + } + validate_id("operation ID", required_template_string(operation, "id")?)?; + validate_semver( + "operation version", + required_template_string(operation, "version")?, + )?; + + required_object(object, "parameters")?; + let runbook = required_object(object, "runbook")?; + if runbook.len() != 3 { + return Err("ops request runbook must contain id, version, and phase".into()); + } + validate_id("runbook ID", required_template_string(runbook, "id")?)?; + validate_semver( + "runbook version", + required_template_string(runbook, "version")?, + )?; + validate_id("runbook phase", required_template_string(runbook, "phase")?)?; + Ok(()) +} + fn required_template_string<'a>( object: &'a serde_json::Map, key: &str, @@ -624,7 +940,8 @@ fn verified_runbook_content(template: &Value) -> Result { .get("runbookDocument") .and_then(Value::as_str) .ok_or("ops request template requires runbookDocument")?; - let actual = exact_runbook_content_sha256(relative)?; + let bytes = exact_runbook_bytes(relative)?; + let actual = runbook_content_digest(&bytes); let declared = object .get("runbookContentSha256") .and_then(Value::as_str) @@ -632,6 +949,11 @@ fn verified_runbook_content(template: &Value) -> Result { if declared != actual { return Err("runbookContentSha256 does not match the exact Markdown runbook bytes".into()); } + if let Some(target) = canonical_runbook_target(&bytes)? { + if object.get("target") != Some(&target) { + return Err("ops request target does not match the exact Markdown runbook bytes".into()); + } + } Ok(actual) } @@ -640,10 +962,6 @@ fn exact_runbook_content_sha256(relative: &str) -> Result { Ok(runbook_content_digest(&bytes)) } -fn exact_runbook_target(relative: &str) -> Result, String> { - canonical_runbook_target(&exact_runbook_bytes(relative)?) -} - fn canonical_runbook_target(bytes: &[u8]) -> Result, String> { const PREFIX: &str = "- Set `target` to `"; const SUFFIX: &str = "`."; @@ -690,12 +1008,10 @@ fn exact_runbook_bytes(relative: &str) -> Result, String> { if !document.starts_with(&runbooks_root) { return Err("runbookDocument must resolve inside the framework runbooks directory".into()); } - let metadata = - fs::metadata(&document).map_err(|error| format!("inspect runbook document: {error}"))?; - if !metadata.is_file() || metadata.len() == 0 || metadata.len() > 1_048_576 { + let (bytes, metadata) = read_bounded_file(&document, MAX_RUNBOOK_BYTES, true)?; + if metadata.len() == 0 { return Err("runbook document must be a regular file between 1 byte and 1 MiB".into()); } - let bytes = fs::read(&document).map_err(|error| format!("read runbook document: {error}"))?; Ok(bytes) } @@ -780,12 +1096,6 @@ fn call_prod_mcp(permit: &str) -> Result { if let Some(error) = result.get("error") { return Err(format!("prod-mcp execution failed: {error}")); } - if result.pointer("/result/isError").and_then(Value::as_bool) == Some(true) { - return Err(format!( - "prod-mcp rejected the operation: {}", - result["result"]["structuredContent"] - )); - } Ok(result) } @@ -1107,8 +1417,9 @@ mod tests { use super::{ base64_decode, base64url_encode, build_request, canonical, curl_command, ecdsa_der_to_raw, parse_mcp_body, private_temp_path, review_binding_marker, review_binding_matches, - review_binding_value, review_evidence_is_bound, reviewer_accepted, runbook_content_digest, - validate_request_template, write_mcp_headers, TrustedApproval, + review_binding_value, review_evidence_is_bound, + reviewer_accepted, runbook_content_digest, validate_request_template, + write_mcp_headers, TrustedApproval, }; use chrono::{TimeZone, Utc}; use serde_json::json; @@ -1316,4 +1627,5 @@ mod tests { let _ = fs::remove_file(request_headers); let _ = fs::remove_file(response_headers); } + } diff --git a/src/runtime.rs b/src/runtime.rs index b5c114b..5bbf5a9 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -253,7 +253,8 @@ pub fn role_agent_exec(args: &[String]) -> Result { } let public_output = dir.join("last-message.txt"); let trace_dir = cfg.logs.join("agents").join(name); - let resume_session = restored + let resume_session = (restored + && env::var("MULTIAGENT_FORCE_FRESH_CONTEXT").as_deref() != Ok("1")) .then(|| native_resume_session(&trace_dir)) .flatten(); let executable = env::current_exe().map_err(io_error("resolve multiagent executable"))?; @@ -1747,18 +1748,22 @@ fn reject_additional_ops_identity(state: &Path, name: &str, role: &str) -> Resul Ok(()) } -fn reviewed_ops_reviewer_instruction(request_file: &Path, request: &str, binding: &str) -> String { +fn reviewed_ops_reviewer_instruction( + request_file: &Path, + descriptor: &str, + binding: &str, +) -> String { format!( - "Independently review the immutable ops request below against its stated goal, operation, target, parameters, and certified runbook. Do not modify or execute it. If and only if it is acceptable, end with an accepted verdict and reproduce the binding marker exactly. Otherwise reject it with concrete findings.\n\nrequest-path: {}\n{}\n\nimmutable-request:\n{}", + "Independently review the supervisor-owned immutable ops request identified by the bounded artifact descriptor below. Read that exact artifact, compare it with its stated goal, operation, target, parameters, and certified runbook, and do not modify or execute it. If and only if it is acceptable, end with an accepted verdict and reproduce the binding marker exactly. Otherwise reject it with concrete findings.\n\nrequest-path: {}\n{}\n\nimmutable-request-descriptor:\n{}", request_file.display(), binding, - request + descriptor ) } fn reviewed_ops_execute_instruction(request_file: &Path, reviewer: &str) -> String { format!( - "Continue the same runbook with the independently reviewed immutable request. Execute exactly:\n\nmultiagent ops execute --request-file {} --reviewer {}\n\nInspect the persisted receipt, continue the runbook if more operations are needed, and report the result or exact blocker. Do not create a replacement ops identity.", + "Continue the same runbook with the independently reviewed immutable request. Execute exactly:\n\nmultiagent ops execute --request-file {} --reviewer {}\n\nInspect the persisted structured outcome and decide from the runbook whether to stop, escalate, or propose another distinct reviewed operation. Never execute the same immutable request twice. Report the result or exact blocker. Do not create a replacement ops identity.", shell_escape(&request_file.display().to_string()), shell_escape(reviewer) ) @@ -1822,10 +1827,12 @@ fn reviewed_ops_cycle(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String "reviewed ops request must belong to ops identity {ops_name}" )); } - let request = - fs::read_to_string(&request_file).map_err(io_error("read reviewed ops request"))?; - let binding = crate::prod_ops::review_binding_for_request(&request_file)?; - let reviewer_instruction = reviewed_ops_reviewer_instruction(&request_file, &request, &binding); + let published = crate::prod_ops::publish_bound_request(&cfg.state, &request_file)?; + let request_file = published.path(); + let descriptor = published.descriptor_json()?; + let binding = crate::prod_ops::review_binding_for_request(request_file)?; + let reviewer_instruction = + reviewed_ops_reviewer_instruction(&request_file, &descriptor, &binding); spawn( cfg, &[ @@ -1848,14 +1855,15 @@ fn reviewed_ops_cycle(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String )); } finalize(cfg, std::slice::from_ref(&reviewer))?; - crate::prod_ops::preflight_reviewed_request(&request_file, &reviewer)?; + crate::prod_ops::preflight_reviewed_request(request_file, &reviewer)?; - let execute_instruction = reviewed_ops_execute_instruction(&request_file, &reviewer); + let execute_instruction = reviewed_ops_execute_instruction(request_file, &reviewer); restore( cfg, &[ ops_name.to_string(), "--force".into(), + "--fresh-context".into(), "--instruction".into(), execute_instruction, ], @@ -2110,6 +2118,7 @@ fn restore(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { .ok_or_else(|| "restore requires NAME".to_string())?; validate_name(name)?; let mut force = false; + let mut fresh_context = false; let mut follow_up = String::new(); let mut follow_up_file = None::; let mut index = 1; @@ -2119,6 +2128,10 @@ fn restore(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { force = true; index += 1; } + "--fresh-context" => { + fresh_context = true; + index += 1; + } "--instruction" => { follow_up = required_value(args, index, "restore --instruction")?.to_string(); index += 2; @@ -2185,23 +2198,31 @@ fn restore(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { if plan.window == "open" && !force { return Err(format!("subagent window already exists: {name}")); } - if !has_recovery_context(&dir) { + if !fresh_context && !has_recovery_context(&dir) { return Err(format!("no captured context to restore: {name}")); } if plan.window == "open" { tmux_checked(&["kill-window", "-t", &format!("{}:{name}", cfg.session)])?; } - let mut instruction = format!( - "You are a restored long-running subagent.\n\nRestoration details:\n- Subagent name: {name}\n- Prior persisted status: {}\n- Persisted state directory: {}\n- This is a fresh tmux window after an orchestrator/session recovery.\n- Do not delete, overwrite, or reset prior memory in the state directory.\n- Read the prior context below, continue only if the assignment is still valid, and report progress/final status in this tmux window.\n- If the prior state shows completion, intentional stop, stale instructions, or a blocker that needs orchestrator/user input, stop and state what you need instead of guessing.\n\nConcise prior context:\n{}\n", - plan.status, - dir.display(), - recovery_text(&dir) - ); - if let Some(context) = implementation_context(cfg, name)? { - instruction.push_str("\n## Approved Implementation Context\n\n"); - instruction.push_str( - &fs::read_to_string(context).map_err(io_error("read implementation context"))?, - ); + let mut instruction = if fresh_context { + format!( + "Continue as the existing logical subagent {name} with the same role, UID, and supervisor-mediated authority. This invocation intentionally starts with a fresh provider model context: do not reconstruct or request prior pane text, transcripts, final messages, or provider output. Use only the typed supervisor-owned state and immutable artifact descriptors supplied by the current instruction.\n" + ) + } else { + format!( + "You are a restored long-running subagent.\n\nRestoration details:\n- Subagent name: {name}\n- Prior persisted status: {}\n- Persisted state directory: {}\n- This is a fresh tmux window after an orchestrator/session recovery.\n- Do not delete, overwrite, or reset prior memory in the state directory.\n- Read the prior context below, continue only if the assignment is still valid, and report progress/final status in this tmux window.\n- If the prior state shows completion, intentional stop, stale instructions, or a blocker that needs orchestrator/user input, stop and state what you need instead of guessing.\n\nConcise prior context:\n{}\n", + plan.status, + dir.display(), + recovery_text(&dir) + ) + }; + if !fresh_context { + if let Some(context) = implementation_context(cfg, name)? { + instruction.push_str("\n## Approved Implementation Context\n\n"); + instruction.push_str( + &fs::read_to_string(context).map_err(io_error("read implementation context"))?, + ); + } } if !follow_up.trim().is_empty() { instruction.push_str("\n## Supervisor Follow-up\n\n"); @@ -2269,13 +2290,20 @@ fn restore(cfg: &RuntimeConfig, args: &[String]) -> Result<(), String> { if !cfg.headless(&cli) { return Err("UID role isolation requires a headless coding-agent backend".into()); } - format!( + let command = format!( "{} role-agent-exec {} --restore", shell_escape(&executable.display().to_string()), shell_escape(name) - ) + ); + if fresh_context { + format!("MULTIAGENT_FORCE_FRESH_CONTEXT=1 {command}") + } else { + command + } } else { - let resume_session = native_resume_session(&trace_dir); + let resume_session = (!fresh_context) + .then(|| native_resume_session(&trace_dir)) + .flatten(); let command = if cfg.headless(&cli) { build_agent_runner_command( &executable, @@ -2924,6 +2952,9 @@ fn build_agent_runner_command( } fn native_resume_session(trace_dir: &Path) -> Option { + if env::var("MULTIAGENT_FORCE_FRESH_CONTEXT").as_deref() == Ok("1") { + return None; + } if env::var("MULTIAGENT_NATIVE_RESUME").as_deref() != Ok("1") { return None; } @@ -4128,11 +4159,12 @@ mod tests { let request = Path::new("/state/logs/agents/ops-primary/request.json"); let review = reviewed_ops_reviewer_instruction( request, - "{\"operation\":{\"id\":\"provider.read\"}}", + "{\"path\":\"/state/request.json\",\"digest\":\"abc\",\"bytes\":42,\"mediaType\":\"application/json\",\"truncated\":false}", "review-binding-sha256=abc", ); assert!(review.contains("review-binding-sha256=abc")); - assert!(review.contains("provider.read")); + assert!(review.contains("immutable-request-descriptor")); + assert!(!review.contains("provider.read")); assert!(!review.contains("Slack")); assert!(!review.contains("Grafana")); @@ -4140,6 +4172,8 @@ mod tests { assert!(execute.contains( "multiagent ops execute --request-file /state/logs/agents/ops-primary/request.json --reviewer ops-reviewer-01" )); + assert!(execute.contains("decide from the runbook")); + assert!(execute.contains("Never execute the same immutable request twice")); } #[cfg(unix)] diff --git a/src/supervisor.rs b/src/supervisor.rs index fecb73d..ae792ca 100644 --- a/src/supervisor.rs +++ b/src/supervisor.rs @@ -1,4 +1,5 @@ use crate::{authority::AuthorityRequest, config, state::read_env as read_env_file}; +#[cfg(target_os = "linux")] use chrono::{SecondsFormat, Utc}; use serde::{Deserialize, Serialize}; use sha2::{Digest, Sha256}; @@ -22,6 +23,10 @@ use std::os::unix::net::UnixStream; const SERVER_CHILD_ENV: &str = "MULTIAGENT_AUTHORITY_SERVER_CHILD"; #[cfg(target_os = "linux")] +const MAX_AUTHORITY_REQUEST_BYTES: u64 = 128 * 1024; +#[cfg(target_os = "linux")] +const MAX_AUTHORITY_OUTPUT_BYTES: usize = 64 * 1024; +#[cfg(target_os = "linux")] const AUTHORITY_REGISTRY: &str = "/run/multiagent/authority-state-10001"; #[cfg(target_os = "linux")] const CONTROL_DIRECTORIES: &[&str] = &[ @@ -713,10 +718,24 @@ fn serve_connection(stream: &mut UnixStream) -> Result { return Ok(false); } let mut bytes = Vec::new(); - if let Err(error) = stream.read_to_end(&mut bytes) { + if let Err(error) = Read::take(&mut *stream, MAX_AUTHORITY_REQUEST_BYTES + 1) + .read_to_end(&mut bytes) + { eprintln!("authority supervisor: read request: {error}"); return Ok(false); } + if bytes.len() as u64 > MAX_AUTHORITY_REQUEST_BYTES { + let _ = write_response( + stream, + &Response { + code: 1, + stdout: String::new(), + stderr: "authority supervisor: request exceeds the configured byte limit\n" + .into(), + }, + ); + return Ok(false); + } let request: AuthorityRequest = match serde_json::from_slice(&bytes) { Ok(request) => request, Err(error) => { @@ -756,7 +775,7 @@ fn serve_connection(stream: &mut UnixStream) -> Result { ); return Ok(false); } - let response = execute(request)?; + let response = execute(request, peer_uid)?; if let Err(error) = write_response(stream, &response) { eprintln!("authority supervisor: {error}"); } @@ -764,24 +783,74 @@ fn serve_connection(stream: &mut UnixStream) -> Result { } #[cfg(target_os = "linux")] -fn execute(request: AuthorityRequest) -> Result { +fn execute(request: AuthorityRequest, peer_uid: u32) -> Result { let executable = env::current_exe().map_err(|error| format!("resolve authority executable: {error}"))?; let (command, args) = request.into_cli(); - let output = Command::new(executable) + let mut child = Command::new(executable) .arg(command) .args(args) .env(SERVER_CHILD_ENV, "1") + .env("MULTIAGENT_AUTHORITY_CALLER_UID", peer_uid.to_string()) .stdin(Stdio::null()) - .output() + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() .map_err(|error| format!("execute authority transaction: {error}"))?; + let stdout = child + .stdout + .take() + .ok_or("authority transaction stdout was not piped")?; + let stderr = child + .stderr + .take() + .ok_or("authority transaction stderr was not piped")?; + let stdout_reader = thread::spawn(move || read_bounded_output(stdout)); + let stderr_reader = thread::spawn(move || read_bounded_output(stderr)); + let status = child + .wait() + .map_err(|error| format!("wait for authority transaction: {error}"))?; + let (stdout, stdout_truncated) = stdout_reader + .join() + .map_err(|_| "authority stdout reader panicked".to_string())??; + let (stderr, stderr_truncated) = stderr_reader + .join() + .map_err(|_| "authority stderr reader panicked".to_string())??; + if stdout_truncated || stderr_truncated { + return Ok(Response { + code: 1, + stdout: String::new(), + stderr: "authority supervisor: transaction output exceeds the configured byte limit\n" + .into(), + }); + } Ok(Response { - code: output.status.code().unwrap_or(1), - stdout: String::from_utf8_lossy(&output.stdout).into_owned(), - stderr: String::from_utf8_lossy(&output.stderr).into_owned(), + code: status.code().unwrap_or(1), + stdout: String::from_utf8_lossy(&stdout).into_owned(), + stderr: String::from_utf8_lossy(&stderr).into_owned(), }) } +#[cfg(target_os = "linux")] +fn read_bounded_output(mut stream: impl Read) -> Result<(Vec, bool), String> { + let mut retained = Vec::with_capacity(MAX_AUTHORITY_OUTPUT_BYTES); + let mut buffer = [0_u8; 8192]; + let mut truncated = false; + loop { + let count = stream + .read(&mut buffer) + .map_err(|error| format!("read authority transaction output: {error}"))?; + if count == 0 { + break; + } + let remaining = MAX_AUTHORITY_OUTPUT_BYTES.saturating_sub(retained.len()); + let keep = remaining.min(count); + retained.extend_from_slice(&buffer[..keep]); + truncated |= keep < count; + } + Ok((retained, truncated)) +} + #[cfg(target_os = "linux")] fn write_response(stream: &mut UnixStream, response: &Response) -> Result<(), String> { let bytes = serde_json::to_vec(response) diff --git a/src/workflow.rs b/src/workflow.rs index a3dcb8c..deca437 100644 --- a/src/workflow.rs +++ b/src/workflow.rs @@ -56,6 +56,7 @@ const USAGE: &str = r#"Usage: multiagent workflow init WORKFLOW_ID multiagent workflow init-or-resume WORKFLOW_ID --resume 0|1 multiagent workflow status WORKFLOW_ID + multiagent workflow context WORKFLOW_ID multiagent workflow contract-register WORKFLOW_ID --scout NAME multiagent workflow prepare-implementation WORKFLOW_ID --decision-id ID --plan-id ID --decision-revision REV --implementation-context PATH --authority-review ID multiagent workflow transition WORKFLOW_ID PHASE [--diff-hash HASH] @@ -81,6 +82,7 @@ pub fn run(args: &[String]) -> Result<(), String> { "init" => initialize(&args[1..], false), "init-or-resume" => init_or_resume(&args[1..]), "status" => status(&args[1..]), + "context" => context(&args[1..]), "contract-register" => register_contract(&args[1..]), "prepare-implementation" => prepare(&args[1..]), "transition" => transition(&args[1..]), @@ -453,6 +455,132 @@ fn status(args: &[String]) -> Result<(), String> { Ok(()) } +fn context(args: &[String]) -> Result<(), String> { + const MAX_CONTEXT_BYTES: usize = 16 * 1024; + const MAX_IDENTITIES: usize = 64; + + let id = one_id("context", args)?; + require_orchestrator_context_caller()?; + let store = Store::configured()?; + let p = store.paths(id)?; + let state = read_env(&p.state, id)?; + validate_original_task(&state)?; + + let task_path = state_value(&state, "original_task"); + if task_path.len() > 1024 { + return Err("original task artifact path exceeds context bound".into()); + } + let task_bytes = fs::metadata(task_path) + .map_err(|error| format!("inspect original task artifact: {error}"))? + .len(); + let identities = typed_identity_context(&store.state_dir, MAX_IDENTITIES)?; + let value = serde_json::json!({ + "apiVersion": "multiagent.moveindustries.io/v1", + "kind": "WorkflowContext", + "workflowId": id, + "phase": bounded_state_label(state_value(&state, "phase"), "phase")?, + "iteration": bounded_state_label(state_value(&state, "iteration"), "iteration")?, + "stateRevision": bounded_state_label(state_value(&state, "decision_revision"), "state revision")?, + "originalTask": { + "path": task_path, + "sha256": state_value(&state, "original_task_sha256"), + "bytes": task_bytes, + "mediaType": "text/plain", + "truncated": false + }, + "activeTodoCount": read_todos(&p.todos)?.iter().filter(|row| active(row.get(4))).count(), + "reviewCount": read_reviews(&p.reviews)?.len(), + "identities": identities + }); + let encoded = serde_json::to_string(&value) + .map_err(|error| format!("encode workflow context: {error}"))?; + if encoded.len() > MAX_CONTEXT_BYTES { + return Err(format!( + "workflow context exceeds strict {MAX_CONTEXT_BYTES}-byte bound" + )); + } + println!("{encoded}"); + Ok(()) +} + +fn require_orchestrator_context_caller() -> Result<(), String> { + #[cfg(target_os = "linux")] + { + let effective_caller = unsafe { libc::geteuid() }; + if effective_caller != config::ORCHESTRATOR_UID { + return Err("workflow context is available only to the orchestrator role".into()); + } + } + Ok(()) +} + +fn typed_identity_context( + state_dir: &Path, + max_identities: usize, +) -> Result, String> { + let root = state_dir.join("subagents"); + if !root.is_dir() { + return Ok(Vec::new()); + } + let mut entries = fs::read_dir(&root) + .map_err(|error| format!("list supervisor identity metadata: {error}"))? + .collect::, _>>() + .map_err(|error| format!("read supervisor identity metadata: {error}"))?; + entries.sort_by_key(|entry| entry.file_name()); + if entries.len() > max_identities { + return Err(format!( + "workflow has more than {max_identities} identities; refusing an incomplete context" + )); + } + let mut identities = Vec::with_capacity(entries.len()); + for entry in entries { + if !entry + .file_type() + .map_err(|error| format!("inspect supervisor identity metadata: {error}"))? + .is_dir() + { + continue; + } + let name = entry.file_name().to_string_lossy().to_string(); + valid_id("subagent identity", &name)?; + let metadata = read_env_optional(&entry.path().join("meta.env"))?; + let role = metadata.get("role").map(String::as_str).unwrap_or("unknown"); + let role = bounded_identity_label(role, "identity role")?; + let raw_status = fs::read_to_string(entry.path().join("status")) + .unwrap_or_else(|_| "unknown".into()); + let status = match raw_status.trim() { + "starting" | "running" | "restoring" | "exited" | "done" | "blocked" + | "stopped" | "killed" | "finalized" => raw_status.trim(), + _ => "unknown", + }; + identities.push(serde_json::json!({ + "name": name, + "role": role, + "status": status + })); + } + Ok(identities) +} + +fn bounded_state_label<'a>(value: &'a str, label: &str) -> Result<&'a str, String> { + if value.len() > 128 || value.chars().any(char::is_control) { + return Err(format!("{label} exceeds workflow context bounds")); + } + Ok(value) +} + +fn bounded_identity_label<'a>(value: &'a str, label: &str) -> Result<&'a str, String> { + if value.is_empty() + || value.len() > 64 + || !value + .chars() + .all(|c| c.is_ascii_alphanumeric() || matches!(c, '_' | '-')) + { + return Err(format!("invalid {label} in supervisor metadata")); + } + Ok(value) +} + fn prepare(args: &[String]) -> Result<(), String> { if args.is_empty() { return Err("prepare-implementation requires WORKFLOW_ID".into());