Move pressure smoke to Rust

This commit is contained in:
Meta Vens 2026-05-06 09:48:25 +01:00
commit a76a35f91d
8 changed files with 305 additions and 130 deletions

View file

@ -267,6 +267,7 @@ pub struct AppServerClient {
stdin: ChildStdin,
rx: mpsc::Receiver<Value>,
transcript: Arc<Mutex<File>>,
notifications: Arc<Mutex<Vec<Value>>>,
next_id: u64,
}
@ -327,6 +328,8 @@ impl AppServerClient {
.ok_or_else(|| anyhow!("app-server stderr unavailable"))?;
let (tx, rx) = mpsc::channel();
let transcript_for_stdout = Arc::clone(&transcript);
let notifications = Arc::new(Mutex::new(Vec::new()));
let notifications_for_stdout = Arc::clone(&notifications);
thread::spawn(move || {
for line in BufReader::new(stdout).lines().map_while(Result::ok) {
if line.trim().is_empty() {
@ -336,6 +339,12 @@ impl AppServerClient {
|error| json!({"_decode_error": error.to_string(), "raw": line}),
);
record(&transcript_for_stdout, "received", &message);
if message.get("method").is_some()
&& message.get("id").is_none()
&& let Ok(mut notifications) = notifications_for_stdout.lock()
{
notifications.push(message.clone());
}
let _ = tx.send(message);
}
});
@ -351,6 +360,7 @@ impl AppServerClient {
stdin,
rx,
transcript,
notifications,
next_id: 1,
})
}
@ -426,6 +436,50 @@ impl AppServerClient {
}
Err(anyhow!("timed out waiting for response {request_id}"))
}
pub fn notification_count(&self, method: &str, start_index: usize) -> usize {
self.notifications
.lock()
.ok()
.map(|notifications| {
notifications
.iter()
.skip(start_index)
.filter(|message| message.get("method").and_then(Value::as_str) == Some(method))
.count()
})
.unwrap_or(0)
}
pub fn notification_len(&self) -> usize {
self.notifications
.lock()
.map(|notifications| notifications.len())
.unwrap_or(0)
}
pub fn require_no_notification(
&mut self,
method: &str,
start_index: usize,
timeout: Duration,
) -> Result<()> {
let deadline = Instant::now() + timeout;
while Instant::now() < deadline {
if let Some(status) = self.child.try_wait()? {
return Err(anyhow!(
"app-server exited with {} while checking notification {}",
status,
method
));
}
if self.notification_count(method, start_index) > 0 {
return Err(anyhow!("unexpected notification {method}"));
}
thread::sleep(Duration::from_millis(100));
}
Ok(())
}
}
impl Drop for AppServerClient {

View file

@ -0,0 +1,245 @@
use anyhow::Context;
use anyhow::Result;
use anyhow::anyhow;
use serde_json::Value;
use serde_json::json;
use std::env;
use std::fs;
use std::path::{Path, PathBuf};
use std::time::Duration;
#[allow(dead_code)]
#[path = "epiphany-mvp-status.rs"]
mod status_cli;
const DEFAULT_APP_SERVER: &str = r"C:\Users\Meta\.cargo-target-codex\debug\codex-app-server.exe";
fn main() -> Result<()> {
let args = Args::parse()?;
let result = run_smoke(&args)?;
println!("{}", serde_json::to_string_pretty(&result)?);
Ok(())
}
#[derive(Debug)]
struct Args {
app_server: PathBuf,
codex_home: PathBuf,
result: PathBuf,
transcript: PathBuf,
stderr: PathBuf,
}
impl Args {
fn parse() -> Result<Self> {
let root = env::current_dir().context("failed to resolve current dir")?;
let mut parsed = Self {
app_server: PathBuf::from(DEFAULT_APP_SERVER),
codex_home: root
.join(".epiphany-smoke")
.join("phase6-pressure-codex-home"),
result: root
.join(".epiphany-smoke")
.join("phase6-pressure-smoke-result.json"),
transcript: root
.join(".epiphany-smoke")
.join("phase6-pressure-smoke-transcript.jsonl"),
stderr: root
.join(".epiphany-smoke")
.join("phase6-pressure-smoke-server.stderr.log"),
};
let mut args = env::args().skip(1);
while let Some(arg) = args.next() {
match arg.as_str() {
"--app-server" => parsed.app_server = take_path(&mut args, "--app-server")?,
"--codex-home" => parsed.codex_home = take_path(&mut args, "--codex-home")?,
"--result" => parsed.result = take_path(&mut args, "--result")?,
"--transcript" => parsed.transcript = take_path(&mut args, "--transcript")?,
"--stderr" => parsed.stderr = take_path(&mut args, "--stderr")?,
other => return Err(anyhow!("unknown argument: {other}")),
}
}
Ok(parsed)
}
}
fn run_smoke(args: &Args) -> Result<Value> {
let root = env::current_dir().context("failed to resolve current dir")?;
let app_server = status_cli::absolute_path(&args.app_server)?;
if !app_server.exists() {
return Err(anyhow!(
"codex app-server binary not found: {}",
app_server.display()
));
}
let codex_home = status_cli::absolute_path(&args.codex_home)?;
let result_path = status_cli::absolute_path(&args.result)?;
let transcript_path = status_cli::absolute_path(&args.transcript)?;
let stderr_path = status_cli::absolute_path(&args.stderr)?;
reset_smoke_paths(
&root,
&[
codex_home.clone(),
result_path.clone(),
transcript_path.clone(),
stderr_path.clone(),
],
)?;
fs::create_dir_all(&codex_home)
.with_context(|| format!("failed to create {}", codex_home.display()))?;
let mut client = status_cli::AppServerClient::start(
&app_server,
&codex_home,
&transcript_path,
&stderr_path,
)?;
client.send(
"initialize",
Some(json!({
"clientInfo": {
"name": "epiphany-phase6-pressure-smoke",
"title": "Epiphany Phase 6 Pressure Smoke",
"version": "0.1.0",
},
"capabilities": {"experimentalApi": true},
})),
true,
)?;
client.send("initialized", None, false)?;
let started = client.send(
"thread/start",
Some(json!({"cwd": root.join("epiphany-core"), "ephemeral": true})),
true,
)?;
let thread_id = started
.pointer("/thread/id")
.and_then(Value::as_str)
.ok_or_else(|| anyhow!("thread/start response missing thread.id"))?
.to_string();
let notification_start = client.notification_len();
let response = client.send(
"thread/epiphany/pressure",
Some(json!({"threadId": thread_id})),
true,
)?;
require(
response.get("threadId").and_then(Value::as_str) == Some(&thread_id),
"pressure response should echo thread id",
)?;
assert_unknown_pressure(&response)?;
client.require_no_notification(
"thread/epiphany/stateUpdated",
notification_start,
Duration::from_secs(1),
)?;
let final_read = client.send(
"thread/read",
Some(json!({"threadId": thread_id, "includeTurns": false})),
true,
)?;
require(
final_read.pointer("/thread/epiphanyState").is_none(),
"pressure reflection should not create Epiphany state",
)?;
let result = json!({
"threadId": thread_id,
"codexHome": codex_home,
"source": response["source"],
"status": response["pressure"]["status"],
"level": response["pressure"]["level"],
"basis": response["pressure"]["basis"],
"shouldPrepareCompaction": response["pressure"]["shouldPrepareCompaction"],
"stateUpdatedNotificationCount": client.notification_count(
"thread/epiphany/stateUpdated",
notification_start,
),
"finalReadHasEpiphanyState": final_read.pointer("/thread/epiphanyState").is_some(),
});
write_json(&result_path, &result)?;
Ok(result)
}
fn assert_unknown_pressure(response: &Value) -> Result<()> {
require(
response.get("source").and_then(Value::as_str) == Some("live"),
"fresh pressure should report live source",
)?;
let pressure = response
.get("pressure")
.ok_or_else(|| anyhow!("pressure response missing pressure object"))?;
require(
pressure.get("status").and_then(Value::as_str) == Some("unknown"),
"fresh pressure should be unknown",
)?;
require(
pressure.get("level").and_then(Value::as_str) == Some("unknown"),
"fresh pressure level should be unknown",
)?;
require(
pressure.get("basis").and_then(Value::as_str) == Some("unknown"),
"fresh pressure basis should be unknown",
)?;
require(
pressure
.get("shouldPrepareCompaction")
.and_then(Value::as_bool)
== Some(false),
"unknown pressure must not recommend compaction prep",
)?;
require(
pressure.get("usedTokens").is_none(),
"fresh pressure should not invent token usage",
)
}
fn reset_smoke_paths(root: &Path, paths: &[PathBuf]) -> Result<()> {
let smoke_root = root.join(".epiphany-smoke").canonicalize().or_else(|_| {
let smoke_root = root.join(".epiphany-smoke");
fs::create_dir_all(&smoke_root)?;
smoke_root.canonicalize()
})?;
for path in paths {
if !path.exists() {
continue;
}
let resolved = path.canonicalize()?;
if resolved == smoke_root || !resolved.starts_with(&smoke_root) {
return Err(anyhow!(
"refusing to delete non-smoke path: {}",
path.display()
));
}
if path.is_dir() {
fs::remove_dir_all(path)?;
} else {
fs::remove_file(path)?;
}
}
Ok(())
}
fn write_json(path: &Path, value: &Value) -> Result<()> {
if let Some(parent) = path.parent() {
fs::create_dir_all(parent)?;
}
fs::write(path, format!("{}\n", serde_json::to_string_pretty(value)?))?;
Ok(())
}
fn require(condition: bool, message: &str) -> Result<()> {
if condition {
Ok(())
} else {
Err(anyhow!("{message}"))
}
}
fn take_path(args: &mut impl Iterator<Item = String>, name: &str) -> Result<PathBuf> {
args.next()
.map(PathBuf::from)
.ok_or_else(|| anyhow!("{name} requires a value"))
}

View file

@ -1507,7 +1507,7 @@ Current tests cover the landed flows at useful seams:
- live app-server smoke for the first Phase 6 context boundary in `tools/epiphany_phase6_context_smoke.py`.
- live app-server smoke for the first Phase 6 graph traversal boundary in `tools/epiphany_phase6_graph_query_smoke.py`.
- live app-server smoke for the first Phase 6 planning boundary in `tools/epiphany_phase6_planning_smoke.py`.
- live app-server smoke for the first Phase 6 pressure boundary in `tools/epiphany_phase6_pressure_smoke.py`.
- live app-server smoke for the first Phase 6 pressure boundary through native `epiphany-phase6-pressure-smoke`.
- live app-server smoke for the first Phase 6 reorientation policy boundary in `tools/epiphany_phase6_reorient_smoke.py`.
- live app-server smoke for the first bounded reorient-launch runtime consumer plus CRRC coordinator recommendations in `tools/epiphany_phase6_reorient_launch_smoke.py`.
- live app-server smoke for the first MVP operator status view through native `epiphany-mvp-status-smoke`.
@ -1531,7 +1531,7 @@ $env:CARGO_TARGET_DIR='C:\Users\Meta\.cargo-target-codex'; cargo test -p codex-a
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' '.\tools\epiphany_phase6_context_smoke.py'
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' '.\tools\epiphany_phase6_graph_query_smoke.py'
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' '.\tools\epiphany_phase6_planning_smoke.py'
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' '.\tools\epiphany_phase6_pressure_smoke.py'
cargo run --manifest-path .\epiphany-core\Cargo.toml --bin epiphany-phase6-pressure-smoke
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' '.\tools\epiphany_phase6_reorient_smoke.py'
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' '.\tools\epiphany_phase6_reorient_launch_smoke.py'
cargo run --manifest-path .\epiphany-core\Cargo.toml --bin epiphany-mvp-status-smoke

View file

@ -70,7 +70,7 @@ The landed machine now has:
- read-only Phase 6 graph traversal through `thread/epiphany/graphQuery`
- live Phase 6 graph traversal smoke coverage in `tools/epiphany_phase6_graph_query_smoke.py`
- read-only Phase 6 context-pressure reflection through `thread/epiphany/pressure`
- live Phase 6 pressure app-server smoke coverage in `tools/epiphany_phase6_pressure_smoke.py`
- live Phase 6 pressure app-server smoke coverage through native `epiphany-phase6-pressure-smoke`
- durable Phase 6 investigation checkpointing in authoritative typed state, prompt rendering, and scene/context reflection
- read-only Phase 6 CRRC reorientation policy through `thread/epiphany/reorient`
- bounded Phase 6 reorient-guided worker launch through `thread/epiphany/reorientLaunch`

View file

@ -151,7 +151,7 @@ The current spine, blessed but not yet finished:
- live graph traversal app-server smoke through `tools/epiphany_phase6_graph_query_smoke.py`
- live planning app-server smoke through `tools/epiphany_phase6_planning_smoke.py`
- live GUI planning adoption smoke through `tools/epiphany_gui_planning_smoke.py`
- live pressure app-server smoke through `tools/epiphany_phase6_pressure_smoke.py`
- live pressure app-server smoke through native `epiphany-phase6-pressure-smoke`
- live reorientation app-server smoke through `tools/epiphany_phase6_reorient_smoke.py`
- live reorient-guided worker launch smoke through `tools/epiphany_phase6_reorient_launch_smoke.py`
- live MVP operator status smoke through native `epiphany-mvp-status-smoke`
@ -246,7 +246,7 @@ For planning state/projection behavior changes, run:
For context-pressure reflection behavior changes, run:
```powershell
& 'C:\Users\Meta\.cache\codex-runtimes\codex-primary-runtime\dependencies\python\python.exe' '.\tools\epiphany_phase6_pressure_smoke.py'
cargo run --manifest-path .\epiphany-core\Cargo.toml --bin epiphany-phase6-pressure-smoke
```
For reorientation policy behavior changes, run:

Binary file not shown.

View file

@ -207,7 +207,7 @@ landed_baseline:
- Use `tools/epiphany_phase6_graph_query_smoke.py` before modifying graph traversal behavior.
- Use `tools/epiphany_phase6_planning_smoke.py` before modifying planning state or the read-only planning projection.
- Use `tools/epiphany_gui_planning_smoke.py` before modifying GUI Objective Draft adoption behavior.
- Use `tools/epiphany_phase6_pressure_smoke.py` before modifying context-pressure reflection behavior.
- Use `cargo run --manifest-path .\epiphany-core\Cargo.toml --bin epiphany-phase6-pressure-smoke` before modifying context-pressure reflection behavior.
- Use `tools/epiphany_phase6_reorient_smoke.py` before modifying bounded CRRC reorientation policy behavior.
- Use `tools/epiphany_phase6_reorient_launch_smoke.py` before modifying the bounded reorient-guided worker launch surface.
- Use `tools/epiphany_phase6_reorient_launch_smoke.py` before modifying the bounded reorient-worker result read-back surface.

View file

@ -1,124 +0,0 @@
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any
from epiphany_phase5_smoke import AppServerClient
from epiphany_phase5_smoke import DEFAULT_APP_SERVER
from epiphany_phase5_smoke import ROOT
from epiphany_phase5_smoke import require
from epiphany_phase5_smoke import reset_smoke_paths
DEFAULT_CODEX_HOME = ROOT / ".epiphany-smoke" / "phase6-pressure-codex-home"
DEFAULT_RESULT = ROOT / ".epiphany-smoke" / "phase6-pressure-smoke-result.json"
DEFAULT_TRANSCRIPT = ROOT / ".epiphany-smoke" / "phase6-pressure-smoke-transcript.jsonl"
DEFAULT_STDERR = ROOT / ".epiphany-smoke" / "phase6-pressure-smoke-server.stderr.log"
def assert_unknown_pressure(response: dict[str, Any]) -> None:
require(response["source"] == "live", "fresh pressure should report live source")
pressure = response["pressure"]
require(pressure["status"] == "unknown", "fresh pressure should be unknown")
require(pressure["level"] == "unknown", "fresh pressure level should be unknown")
require(pressure["basis"] == "unknown", "fresh pressure basis should be unknown")
require(
pressure["shouldPrepareCompaction"] is False,
"unknown pressure must not recommend compaction prep",
)
require(
"usedTokens" not in pressure,
"fresh pressure should not invent token usage",
)
def run_smoke(args: argparse.Namespace) -> dict[str, Any]:
app_server = args.app_server.resolve()
if not app_server.exists():
raise FileNotFoundError(f"codex app-server binary not found: {app_server}")
codex_home = args.codex_home.resolve()
result_path = args.result.resolve()
transcript_path = args.transcript.resolve()
stderr_path = args.stderr.resolve()
reset_smoke_paths(codex_home, result_path, transcript_path, stderr_path)
with AppServerClient(app_server, codex_home, transcript_path, stderr_path) as client:
client.send(
"initialize",
{
"clientInfo": {
"name": "epiphany-phase6-pressure-smoke",
"title": "Epiphany Phase 6 Pressure Smoke",
"version": "0.1.0",
},
"capabilities": {"experimentalApi": True},
},
)
client.send("initialized", expect_response=False)
started = client.send(
"thread/start",
{"cwd": str(ROOT / "epiphany-core"), "ephemeral": True},
)
assert started is not None
thread_id = started["thread"]["id"]
notification_start = len(client.notifications)
response = client.send("thread/epiphany/pressure", {"threadId": thread_id})
assert response is not None
require(response["threadId"] == thread_id, "pressure response should echo thread id")
assert_unknown_pressure(response)
client.require_no_notification(
"thread/epiphany/stateUpdated",
start_index=notification_start,
)
final_read = client.send("thread/read", {"threadId": thread_id, "includeTurns": False})
assert final_read is not None
require(
"epiphanyState" not in final_read["thread"],
"pressure reflection should not create Epiphany state",
)
result = {
"threadId": thread_id,
"codexHome": str(codex_home),
"source": response["source"],
"status": response["pressure"]["status"],
"level": response["pressure"]["level"],
"basis": response["pressure"]["basis"],
"shouldPrepareCompaction": response["pressure"]["shouldPrepareCompaction"],
"stateUpdatedNotificationCount": client.count_notifications(
"thread/epiphany/stateUpdated",
start_index=notification_start,
),
"finalReadHasEpiphanyState": "epiphanyState" in final_read["thread"],
}
result_path.parent.mkdir(parents=True, exist_ok=True)
result_path.write_text(
json.dumps(result, indent=2, ensure_ascii=False) + "\n",
encoding="utf-8",
)
return result
def main() -> int:
parser = argparse.ArgumentParser(
description="Live-smoke the Phase 6 Epiphany pressure reflection surface."
)
parser.add_argument("--app-server", type=Path, default=DEFAULT_APP_SERVER)
parser.add_argument("--codex-home", type=Path, default=DEFAULT_CODEX_HOME)
parser.add_argument("--result", type=Path, default=DEFAULT_RESULT)
parser.add_argument("--transcript", type=Path, default=DEFAULT_TRANSCRIPT)
parser.add_argument("--stderr", type=Path, default=DEFAULT_STDERR)
args = parser.parse_args()
result = run_smoke(args)
print(json.dumps(result, indent=2, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())