Skip to content

Commit 1186c2e

Browse files
fffonionfffonion
committed
fix(tools): align rss file mutation lifecycle
Map patch write-budget and publication_indeterminate envelopes to native, select artifact summaries by content-byte length, and add lifecycle/race proofs for cancellation, replay, CAS, symlink, and commit-failure.
1 parent 834aef4 commit 1186c2e

9 files changed

Lines changed: 1008 additions & 62 deletions

File tree

‎rss/tools/patch.rss‎

Lines changed: 26 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,16 @@ fn fail_host(envelope: map) -> map {
8787
if code == "wrong_type" {
8888
code = "path_denied";
8989
}
90-
fail(code, message, unpublished())
90+
if code == "budget_exceeded" {
91+
if message == "requested write exceeds the configured bound" {
92+
message = "write budget exceeded";
93+
}
94+
}
95+
let mut data: map = unpublished();
96+
if code == "publication_indeterminate" {
97+
data = { publication: "indeterminate" };
98+
}
99+
fail(code, message, data)
91100
}
92101

93102
fn deny_path(message: string) -> map {
@@ -410,6 +419,21 @@ fn fit_content_to_budget(result: map, cap: int) -> map {
410419
fitted
411420
}
412421

422+
fn artifact_summary(id: string, bytes: int, max_output_bytes: int) -> string {
423+
let full: string = "artifact " + id + " (" + int_to_string(bytes) + " bytes)";
424+
let mut summary: string = full;
425+
if utf8_len(full) > max_output_bytes {
426+
summary = "artifact " + id;
427+
if utf8_len(summary) > max_output_bytes {
428+
summary = "artifact";
429+
if utf8_len(summary) > max_output_bytes {
430+
summary = truncate_to_utf8_bytes("artifact", max_output_bytes);
431+
}
432+
}
433+
}
434+
summary
435+
}
436+
413437
fn shrink_result(result: map, max_output_bytes: int, token: string) -> map {
414438
let mut current: map = result;
415439
if encoded_len(current) > max_output_bytes {
@@ -423,14 +447,7 @@ fn shrink_result(result: map, max_output_bytes: int, token: string) -> map {
423447
let len: int = map_int(put, "len", utf8_len(content));
424448
let mut artifacts: array = [];
425449
artifacts[0] = id;
426-
let mut summary: string = "artifact " + id + " (" + int_to_string(len) + " bytes)";
427-
current = succeed(summary, data, true, artifacts);
428-
if encoded_len(current) > max_output_bytes {
429-
current = succeed("artifact " + id, data, true, artifacts);
430-
}
431-
if encoded_len(current) > max_output_bytes {
432-
current = succeed("artifact", data, true, artifacts);
433-
}
450+
current = succeed(artifact_summary(id, len, max_output_bytes), data, true, artifacts);
434451
}
435452
}
436453
if encoded_len(current) > max_output_bytes {

‎rss/tools/write_file.rss‎

Lines changed: 26 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -87,7 +87,16 @@ fn fail_host(envelope: map) -> map {
8787
if code == "wrong_type" {
8888
code = "path_denied";
8989
}
90-
fail(code, message, unpublished())
90+
if code == "budget_exceeded" {
91+
if message == "requested write exceeds the configured bound" {
92+
message = "write budget exceeded";
93+
}
94+
}
95+
let mut data: map = unpublished();
96+
if code == "publication_indeterminate" {
97+
data = { publication: "indeterminate" };
98+
}
99+
fail(code, message, data)
91100
}
92101

93102
fn deny_path(message: string) -> map {
@@ -294,6 +303,21 @@ fn fit_content_to_budget(result: map, cap: int) -> map {
294303
fitted
295304
}
296305

306+
fn artifact_summary(id: string, bytes: int, max_output_bytes: int) -> string {
307+
let full: string = "artifact " + id + " (" + int_to_string(bytes) + " bytes)";
308+
let mut summary: string = full;
309+
if utf8_len(full) > max_output_bytes {
310+
summary = "artifact " + id;
311+
if utf8_len(summary) > max_output_bytes {
312+
summary = "artifact";
313+
if utf8_len(summary) > max_output_bytes {
314+
summary = truncate_to_utf8_bytes("artifact", max_output_bytes);
315+
}
316+
}
317+
}
318+
summary
319+
}
320+
297321
fn shrink_result(result: map, max_output_bytes: int, token: string) -> map {
298322
let mut current: map = result;
299323
if encoded_len(current) > max_output_bytes {
@@ -307,14 +331,7 @@ fn shrink_result(result: map, max_output_bytes: int, token: string) -> map {
307331
let len: int = map_int(put, "len", utf8_len(content));
308332
let mut artifacts: array = [];
309333
artifacts[0] = id;
310-
let mut summary: string = "artifact " + id + " (" + int_to_string(len) + " bytes)";
311-
current = succeed(summary, data, true, artifacts);
312-
if encoded_len(current) > max_output_bytes {
313-
current = succeed("artifact " + id, data, true, artifacts);
314-
}
315-
if encoded_len(current) > max_output_bytes {
316-
current = succeed("artifact", data, true, artifacts);
317-
}
334+
current = succeed(artifact_summary(id, len, max_output_bytes), data, true, artifacts);
318335
}
319336
}
320337
if encoded_len(current) > max_output_bytes {

‎src/capabilities/filesystem.rs‎

Lines changed: 23 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -81,6 +81,8 @@ pub struct FsWrite {
8181
pub staging_cleaned: bool,
8282
}
8383

84+
type BeforeWriteHook = Arc<dyn Fn(&str, &[u8]) -> Result<(), CapabilityError> + Send + Sync>;
85+
8486
/// Confined filesystem capability bound to one lifecycle owner.
8587
#[derive(Clone)]
8688
pub struct FilesystemCapability {
@@ -90,6 +92,7 @@ pub struct FilesystemCapability {
9092
root: Arc<ConfinedFsRoot>,
9193
frozen: Arc<FrozenDir>,
9294
cas_locks: Arc<Mutex<HashMap<String, Arc<Mutex<()>>>>>,
95+
before_write: Arc<Mutex<Option<BeforeWriteHook>>>,
9396
}
9497

9598
impl FilesystemCapability {
@@ -127,9 +130,21 @@ impl FilesystemCapability {
127130
root: Arc::new(root),
128131
frozen: Arc::new(frozen),
129132
cas_locks: Arc::new(Mutex::new(HashMap::new())),
133+
before_write: Arc::new(Mutex::new(None)),
130134
})
131135
}
132136

137+
/// Installs a production-neutral pre-publish hook for tests.
138+
///
139+
/// The hook runs after authorization and the write-size bound, immediately
140+
/// before the per-path CAS lock and atomic publish. It is tool-name-agnostic.
141+
pub fn inject_before_write(&self, hook: BeforeWriteHook) {
142+
*self
143+
.before_write
144+
.lock()
145+
.unwrap_or_else(|poisoned| poisoned.into_inner()) = Some(hook);
146+
}
147+
133148
/// Stats a workspace-relative path without following a leaf symlink.
134149
pub fn metadata(&self, token: &str, path: &str) -> Result<FsMetadata, CapabilityError> {
135150
let _claims = self.authorize(token, CapabilityRisk::Read)?;
@@ -250,6 +265,14 @@ impl FilesystemCapability {
250265
"requested write exceeds the configured bound",
251266
));
252267
}
268+
if let Some(hook) = self
269+
.before_write
270+
.lock()
271+
.unwrap_or_else(|poisoned| poisoned.into_inner())
272+
.clone()
273+
{
274+
hook(path, bytes)?;
275+
}
253276
let lock = self.lock_for(path);
254277
let _guard = lock.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
255278
if expected_hash != "*" {

‎src/lib.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,9 @@ pub use runtime::rss_runner::{
3333
RUN_EPOCH_DEADLINE_TICKS, Result, RunCancellation, RunDeliveryError, RunError, RunEventSink,
3434
RunnerPrepareFault,
3535
};
36-
pub use runtime::{AgentHostBridges, AgentProviderHost, ScriptedProvider, agent_host_catalog};
36+
pub use runtime::{
37+
AgentHostBridges, AgentProviderHost, ControlCheckHook, ScriptedProvider, agent_host_catalog,
38+
};
3739
pub use service::{
3840
AdmitError, AdmitRunRequest, AdmittedRun, AgentService, CleanupOutcome, ProviderCommit,
3941
ProviderCommitOutcome, ProviderPendingDecision, RunHandle,

‎src/runtime/agent_host.rs‎

Lines changed: 19 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -10,9 +10,9 @@ use std::thread;
1010
use std::time::{Duration, Instant};
1111

1212
use rustscript_vm::{
13-
CallOutcome, CallReturn, HostApiBuilder, HostApiCatalog, HostFunctionRegistry,
14-
HostFunctionSchema, HostParamSchema, HostTypeSchema, Value, Vm, VmError, VmResult,
15-
catalog_import_schemas, standard_host_catalog,
13+
CallOutcome, CallReturn, CancellationReason, HostApiBuilder, HostApiCatalog,
14+
HostFunctionRegistry, HostFunctionSchema, HostParamSchema, HostTypeSchema, Value, Vm, VmError,
15+
VmResult, catalog_import_schemas, standard_host_catalog,
1616
};
1717
use serde_json::{Value as JsonValue, json};
1818

@@ -258,6 +258,8 @@ pub trait AgentProviderHost: Send + Sync {
258258
fn call(&self, request: &JsonValue, cancellation: &RunCancellation) -> JsonValue;
259259
}
260260

261+
pub type ControlCheckHook = Arc<dyn Fn(&RunCancellation) + Send + Sync>;
262+
261263
/// Injectable host bridges for one compiled runner.
262264
#[derive(Clone, Default)]
263265
pub struct AgentHostBridges {
@@ -273,6 +275,9 @@ pub struct AgentHostBridges {
273275
pub filesystem: Option<Arc<FilesystemCapability>>,
274276
pub processes: Option<Arc<ProcessCapability>>,
275277
pub artifacts: Option<Arc<ArtifactCapability>>,
278+
/// Optional test hook invoked from `agent::control_check` before reading
279+
/// cancellation/deadline. Production callers leave this unset.
280+
pub control_hook: Option<ControlCheckHook>,
276281
}
277282

278283
/// Per-VM state installed before `run(context)`.
@@ -290,12 +295,21 @@ pub struct AgentHostState {
290295
pub processes: Option<Arc<ProcessCapability>>,
291296
pub artifacts: Option<Arc<ArtifactCapability>>,
292297
pub(crate) leases: Arc<Mutex<HashMap<String, ExecutionLease>>>,
298+
pub(crate) control_hook: Option<ControlCheckHook>,
293299
}
294300

295301
impl AgentHostState {
296302
fn control_error(&self) -> Option<JsonValue> {
297-
if self.cancellation.requested().is_some() {
298-
return Some(typed_fail("cancelled", "run was cancelled"));
303+
if let Some(hook) = &self.control_hook {
304+
hook(&self.cancellation);
305+
}
306+
if let Some(reason) = self.cancellation.requested() {
307+
return Some(match reason {
308+
CancellationReason::Deadline => {
309+
typed_fail("deadline_elapsed", "run deadline elapsed")
310+
}
311+
_ => typed_fail("cancelled", "run was cancelled"),
312+
});
299313
}
300314
if self.cancellation.deadline_passed() {
301315
return Some(typed_fail("deadline_elapsed", "run deadline elapsed"));

‎src/runtime/mod.rs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@ pub(crate) mod agent_host;
44
pub(crate) mod delivery;
55
pub mod rss_runner;
66

7-
pub use agent_host::{AgentHostBridges, AgentProviderHost, ScriptedProvider, agent_host_catalog};
7+
pub use agent_host::{
8+
AgentHostBridges, AgentProviderHost, ControlCheckHook, ScriptedProvider, agent_host_catalog,
9+
};
810
pub use rss_runner::{
911
AgentConfig, AgentError, AgentRunner, MAX_AGENT_SOURCE_BYTES, RUN_EPOCH_CHECK_INTERVAL,
1012
RUN_EPOCH_DEADLINE_TICKS, Result, RunCancellation, RunDeliveryError, RunError, RunEventSink,

‎src/runtime/rss_runner.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -628,6 +628,7 @@ impl AgentRunner {
628628
processes: self.host.processes.clone(),
629629
artifacts: self.host.artifacts.clone(),
630630
leases: Arc::new(Mutex::new(HashMap::new())),
631+
control_hook: self.host.control_hook.clone(),
631632
});
632633
if let Some(cancellation) = cancellation {
633634
vm.set_epoch_check_interval(RUN_EPOCH_CHECK_INTERVAL)

‎src/service.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3314,6 +3314,7 @@ impl AgentService {
33143314
filesystem,
33153315
processes,
33163316
artifacts,
3317+
control_hook: None,
33173318
};
33183319
// One bounded delivery path: the worker blocks on this channel
33193320
// when the delivery task is busy, which pauses invocation polling

0 commit comments

Comments
 (0)