Build a shared worker fleet
Run competing workers against one durable authority, account for every report, and recover a lost acknowledgement without producing a report twice.
On this page
A fleet of workers can share one allowance when a single host owns the spending and the finished work. This application starts separate worker processes that claim report sections from one authoritative host, which holds the grant, the accounting, the operation records, and publication.
Run the fleetThree workers compete for report sections under one allowance
One allowance. Many workers.
Independent worker processes compete for report jobs through one authoritative host. Retain spending and reconcile lost acknowledgements.
Supply a document, choose the sections to produce, and change the concurrency. Each report costs 20 local work credits, so an allowance of 50 funds two reports; raise it to fund more. Enable the lost-acknowledgement case and restart: the report commits before the connection fails, and the restarted host recovers that operation instead of calling the report handler again.
One host owns the allowance, the claims, and the operation records
The workers hold a credential and a claim. Everything that decides spending and completion lives in the host process, so another worker adds contention for claims and never adds budget.
One allowance funds the whole fleet
The host owns the grant and the shared allowance. Each report reserves 20 local work credits before dispatch, so an allowance of 50 funds two reports and the kernel refuses the third. Workers hold a credential and a claim, never a copy of the mission budget. Shared budgets.
Authority and the allowance survive process replacement
The shared host uses the same durable admission runtime as the other applications. Its SQLite state retains the kernel identity and the allowance when the process is replaced. Each worker client holds its assigned credential and claim, never its own copy of the mission budget.
// Read-only tools consume authority too. Include them in durable quota accounting.
kernel.configure_durable_admission(
chio_kernel::admission_operation::DurableAdmissionMode::All,
false,
)?;
authority.attach(&mut kernel)?;
let budget = authority
.local_budget_store()
.context("Local accounting is required")?;
for server in servers {
kernel.register_tool_server(server);
}
write_json(
&path.join("trusted-kernel.json"),
&json!({"public_key":signer.to_hex()}),
)?;
Ok(Self {
kernel: Arc::new(kernel),
signer,
issuer: keypair,
budget,
})
}In the verified run, event 22 restarts the host with the original allowance: 40 of 50 local credits spent and 10 available.
Workers claim jobs at the shared owner
The claim endpoint authenticates the worker, selects an available job, and persists ownership before it returns the job. Every later operation presents that exact claim. The small queue keeps its scheduling policy in view, and Chio’s admission runtime decides which execution and spending proceed.
async fn claim(State(service): State<Arc<Service>>, headers: HeaderMap) -> ApiResult {
let owner = worker(&headers, &service)?;
let mut stored = service.jobs.lock().await;
let mut jobs = stored.clone();
let Some(job) = jobs
.iter_mut()
.find(|job| matches!(job.state, JobState::Queued))
else {
return Ok(Json(json!({"job":null})));
};
job.claim(owner.clone()).map_err(error)?;
let result = job.clone();
service.persist(&jobs).map_err(error)?;
*stored = jobs;
service
.run
.emit(
"fleet.claimed",
&owner,
"Worker claimed a report section",
json!({"operation_id":result.operation_id,"section":result.input["section"]}),
)
.map_err(error)?;
Ok(Json(json!({"job":result})))
}Events 05, 07, and 08 record three workers claiming report sections. Event 11 refuses worker 3’s report once the allowance is spent, with receipt 4ffd0944…615984.
Cancellation is a recorded decision against one job
Cancellation records a decision against the named job and interrupts the running report. A partial draft is never a published report. The operations chapter cancels while the handler is active, so its result shows the adapter’s state after the interruption.
async fn cancel(
State(service): State<Arc<Service>>,
headers: HeaderMap,
Path(id): Path<String>,
) -> ApiResult {
admin(&headers, &service)?;
if let Some(flag) = service.cancellations.lock().await.get(&id) {
flag.store(true, Ordering::SeqCst);
return Ok(Json(json!({"cancellation_requested":true})));
}
let mut stored = service.jobs.lock().await;
let mut jobs = stored.clone();
let job = jobs
.iter_mut()
.find(|job| job.operation_id.to_string() == id)
.ok_or_else(|| error("Unknown operation"))?;
if matches!(job.state, JobState::Queued | JobState::Claimed { .. }) {
job.cancel_before_dispatch().map_err(error)?;
service.persist(&jobs).map_err(error)?;
*stored = jobs;
return Ok(Json(
json!({"state":"cancelled_before_dispatch","receipt_id":null}),
));
}
Err(error(
"No active worker; inspect the operation before attempting recovery",
))
}Reconciliation resolves an operation before any retry
The report adapter records an operation ID with its completed output. Reconciliation asks the adapter whether the effect committed, then joins the result to its retained job and receipt. A known report is recovered and an unproven effect stays unresolved. Each new effectful tool defines its own reconciliation contract.
async fn reconcile(
State(service): State<Arc<Service>>,
headers: HeaderMap,
Path(id): Path<String>,
) -> ApiResult {
admin(&headers, &service)?;
let mut stored = service.jobs.lock().await;
let mut jobs = stored.clone();
let job = jobs
.iter_mut()
.find(|job| job.operation_id.to_string() == id)
.ok_or_else(|| error("Unknown operation"))?;
if service.cancellations.lock().await.contains_key(&id) {
return Err(error(
"Worker is still executing; wait for its terminal state",
));
}
let file = service
.config
.directory
.join("reports")
.join(format!("{id}.json"));
let resolution = if file.exists() {
let artifact: Value =
serde_json::from_slice(&std::fs::read(file).map_err(error)?).map_err(error)?;
let expected =
chio_core::sha256_hex(&chio_core::canonical_json_bytes(&job.input).map_err(error)?);
if artifact["input_sha256"] != expected || artifact["operation_id"] != id {
return Err(error(
"Artifact does not match this operation; leave it unresolved",
));
}
let outcome = job.reconcile_published().map_err(error)?;
json!({"state":"completed","reused_existing_effect":true,"artifact":artifact,"receipt_id":outcome["receipt_id"]})
} else {
json!({
"state":job.label(),
"reused_existing_effect":false,
"recovery":"No published artifact proves completion. Inspect retained admission and partial output; this adapter does not blindly retry uncertain work."
})
};
service.persist(&jobs).map_err(error)?;
*stored = jobs;
service
.run
.emit(
"fleet.reconciled",
"operator",
"Reconciled the report's actual state",
json!({"operation_id":id,"resolution":resolution}),
)
.map_err(error)?;
Ok(Json(resolution))
}Event 23 reconciles operation b9bf2f70…dcbf49 to the receipt the kernel signed in event 17, ff11eb4b…7f52f8.
Spending past the allowance is refused and a restart recovers the reports
The sample input funds two of four reports and restarts the host before reconciliation. The retained run holds these events:
./run --run input.json11 call.denied worker-3
invocation budget exhausted for capability
cap-01a08c22-2f0d-7500-abac-8638fdb0c255
receipt 4ffd0944…615984
15 call.denied worker-3
invocation budget exhausted for capability
cap-01a08c22-2f0d-7500-abac-8638fdb0c255
receipt d3d82131…b1ca76
21 fleet.host_stopped operator
Stopped the host before reconciliation; 40 of 50 local demo credits spent
22 fleet.host_restarted operator
Restart retained the original allowance and report records; 40 of 50 local demo
credits spent
23 fleet.reconciled operator
state completed
receipt ff11eb4b…7f52f8- Spend beyond the shared allowance: refused jobs produce no report.
- Reuse a stale worker claim after host recovery: the host rejects it.
- Lose the response after the report commits: reconciliation returns the original result and receipt without another handler execution.
The downloadable tests assert each effect or its absence: run cargo test --locked -p chio-worker-fleet from the extracted project.
Run the fleet on your machine
Keep the run directory to inspect worker claims, shared accounting, published reports, and recovered operation results.
cd chio-worker-fleet
./runThe archive holds the application, the shared host code, tests, sample input, pinned dependencies, and a README. The launcher builds and runs the Rust source included in this archive. Install Rust 1.94.1 and the README’s native prerequisites; filesystem isolation and operator workloads require the listed Linux host profile. Cargo resolves immutable public dependencies, so a sibling checkout is unnecessary.
For model workers, set OPENROUTER_API_KEY on the local host and select model mode; credentials stay on that host and never enter the docs form. The qualified model is openai/gpt-4.1-mini. Analysis, retrieval, and deterministic worker modes run without a model key. Ctrl-C stops the local web process and its active child work and keeps the run directory.
Download identity and source revision
Download complete project · View public source
c54d2b8c9c337e640777e708697b20f917d32768f3b338da5f2382e3e5cc03f8Add a report section and keep its operation ID
Add a section to the fleet’s report tool and teach the queue which input it needs. Preserve the operation ID at the effect owner. Worker clients move to separate machines once you configure their authenticated transport; the starter runs separate local processes against one central authority.