skills/pbakaus_impeccable
1
0
Fork 0
mirror of https://github.com/pbakaus/impeccable.git synced 2026-09-17 00:26:41 +03:00
Code Issues Packages Projects Releases Wiki Activity
Files
c0a0f8a9ad3a536f80736311b364416e2bfcdead
pbakaus_impeccable/crates/live/src/session.rs
T
Abdul WahabAbdul WahabClaude Fable 5
 e06c152ad2 generate lane: one-shot start, event in hand, mechanical bake 
The generate command's fast lane spent most of its time on agent round
trips, not on the engine. Four engine changes take them out, all behind
the lane's own flags so a plain `live` session is untouched:

- `live-poll --reply <id> done --then-poll` replies and waits for the
  next event in one call; the reply's ack rides along as `_replyAck`.
- `live-generate` collects the session's own generate event into its
  output (`GET /poll?types=generate&id=<sessionId>`, a new id filter
  the parked-poll flush honours too), so the pickup poll is gone.
- `live-generate --boot` runs the lane's boot in-process and reuses a
  running helper; `--dev-url <url>` names the dev server the agent
  already knows and leads the probe; with no page connected the verdict
  is `browser_needed` with the harness's own way to open the page
  (Cursor browser_navigate, Claude Code's Browser pane, Codex --open or
  the user). `--open` launches the system browser only on a harness
  without one: on cursor and claude-code it is ignored unless
  IMPECCABLE_BROWSER or the config's `browser` names a browser, so a
  second window never opens beside the harness's. The served /live.js
  carries the helper-wide bar preference in its prelude.
- The accept of a session the generate verb started (journaled with
  origin "agent", or `--bake`) is baked mechanically: the accepted
  variant's @scope rules are re-anchored on the element's own selector
  and appended to the stylesheet that names it, the wrapper is
  unwrapped, the source verified clean. Knobs, plumbing inside the
  variant, no stylesheet, or a selector the rewrite cannot decide fall
  back to the carbonize block with `bakeSkipped`. `--no-bake` refuses.

Goldens re-recorded for the three help texts and the no-browser case.

Written with AI assistance (Claude).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-09-15 05:45:49 +05:00

1006 lines
36 KiB
Rust
Raw Blame History

//! JS: live/session-store.mjs. Append-only per-session journal
//! (`<id>.jsonl`) with a derived snapshot (`<id>.snapshot.json`), replayed
//! through the same reducer the server uses.
use crate::paths::{legacy_live_sessions_dir, live_sessions_dir, safe_session_id};
use crate::util::{
date_parse_ms, exists, iso_now, js_number, jsp, now_ms, read_dir_names_raw, safe_read, Env,
};
use serde_json::{json, Map, Value};
pub const COMPLETED_SESSION_PHASES: [&str; 2] = ["completed", "discarded"];
pub const GENERATION_FENCED_SESSION_PHASES: [&str; 5] = [
"accept_requested",
"discard_requested",
"carbonize_required",
"completed",
"discarded",
];
const META_JOURNAL_BYTES: &str = "__journalBytes";
const META_NEXT_SEQ: &str = "__nextSeq";
const MOUNT_FAILURE_HISTORY: usize = 5;
pub struct SessionStore {
pub root_dir: String,
pub legacy_root_dir: String,
pub session_id: Option<String>,
}
pub struct DerivedState {
pub snapshot: Map<String, Value>,
pub next_seq: i64,
pub journal_path: String,
pub size: i64,
}
fn journal_path(root: &str, id: &str) -> Result<String, String> {
Ok(jsp::join(&[
root,
&format!("{}.jsonl", safe_session_id(id)?),
]))
}
fn snapshot_path(root: &str, id: &str) -> Result<String, String> {
Ok(jsp::join(&[
root,
&format!("{}.snapshot.json", safe_session_id(id)?),
]))
}
fn file_size(p: &str) -> Option<i64> {
std::fs::metadata(p).ok().map(|m| m.len() as i64)
}
/// JS: createLiveSessionStore({ cwd, sessionId })
pub fn create_live_session_store(cwd: &str, env: &Env, session_id: Option<&str>) -> SessionStore {
let root_dir = live_sessions_dir(cwd, env);
let legacy_root_dir = legacy_live_sessions_dir(cwd, env);
let _ = std::fs::create_dir_all(&root_dir);
SessionStore {
root_dir,
legacy_root_dir,
session_id: session_id.map(String::from),
}
}
impl SessionStore {
fn readable_journal_path(&self, id: &str) -> Result<String, String> {
let primary = journal_path(&self.root_dir, id)?;
if exists(&primary) {
return Ok(primary);
}
let legacy = journal_path(&self.legacy_root_dir, id)?;
if exists(&legacy) {
return Ok(legacy);
}
Ok(primary)
}
/// JS: readState(id, { allowSnapshotFile })
pub fn read_state(&self, id: &str, allow_snapshot_file: bool) -> Result<DerivedState, String> {
let journal = self.readable_journal_path(id)?;
let size = file_size(&journal);
if allow_snapshot_file {
if let Some(sz) = size {
if let Some((snapshot, next_seq)) =
read_snapshot_file(&snapshot_path(&self.root_dir, id)?, id, sz)
{
return Ok(DerivedState {
snapshot,
next_seq,
journal_path: journal,
size: sz,
});
}
}
}
let (snapshot, next_seq) = rebuild_snapshot_from_journal(&journal, id);
Ok(DerivedState {
snapshot,
next_seq,
journal_path: journal,
size: size.unwrap_or(-1),
})
}
fn persist(
&self,
id: &str,
snapshot: &Map<String, Value>,
next_seq: i64,
) -> Result<(), String> {
let sp = snapshot_path(&self.root_dir, id)?;
let journal = self.readable_journal_path(id)?;
let size = file_size(&journal).unwrap_or(-1);
write_snapshot(&sp, snapshot, size, next_seq);
Ok(())
}
/// JS: appendEvent(event). Returns the new snapshot.
pub fn append_event(&self, event: &Value) -> Result<Map<String, Value>, String> {
let normalized = normalize_event(event, self.session_id.as_deref())?;
let id = normalized
.get("id")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let ty = normalized
.get("type")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
let jp = journal_path(&self.root_dir, &id)?;
let legacy = journal_path(&self.legacy_root_dir, &id)?;
if !exists(&jp) && exists(&legacy) {
let _ = std::fs::create_dir_all(jsp::dirname(&jp));
let _ = std::fs::copy(&legacy, &jp);
}
let prior = self.read_state(&id, true)?;
let entry = json!({
"seq": prior.next_seq,
"id": id,
"type": ty,
"ts": iso_now(),
"event": Value::Object(normalized.clone()),
});
{
use std::io::Write;
let _ = std::fs::create_dir_all(jsp::dirname(&jp));
let mut f = std::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&jp)
.map_err(|e| e.to_string())?;
let _ = f.write_all(
format!("{}\n", serde_json::to_string(&entry).unwrap_or_default()).as_bytes(),
);
}
let next = apply_event(&prior.snapshot, &entry);
self.persist(&id, &next, prior.next_seq + 1)?;
Ok(next)
}
/// JS: has(id)
pub fn has(&self, id: &str) -> bool {
if id.is_empty() {
return false;
}
match (
journal_path(&self.root_dir, id),
journal_path(&self.legacy_root_dir, id),
) {
(Ok(a), Ok(b)) => exists(&a) || exists(&b),
_ => false,
}
}
/// JS: getSnapshot(id, { includeCompleted }); None for terminal sessions
/// unless `include_completed`.
pub fn get_snapshot(
&self,
id: &str,
include_completed: bool,
) -> Result<Option<Map<String, Value>>, String> {
let state = self.read_state(id, true)?;
let phase = state
.snapshot
.get("phase")
.and_then(|p| p.as_str())
.unwrap_or("");
if !include_completed && COMPLETED_SESSION_PHASES.contains(&phase) {
return Ok(None);
}
Ok(Some(state.snapshot))
}
/// JS: flush(id)
pub fn flush(&self, id: &str) -> Result<Map<String, Value>, String> {
let state = self.read_state(id, false)?;
self.persist(id, &state.snapshot, state.next_seq)?;
Ok(state.snapshot)
}
/// JS: listActiveSessions(): all `*.jsonl` ids in legacy + primary,
/// sorted, non-terminal.
pub fn list_active_sessions(&self) -> Vec<Map<String, Value>> {
let mut ids: Vec<String> = Vec::new();
for dir in [&self.legacy_root_dir, &self.root_dir] {
if !exists(dir) {
continue;
}
for name in read_dir_names_raw(dir).unwrap_or_default() {
if let Some(id) = name.strip_suffix(".jsonl") {
if !ids.iter().any(|x| x == id) {
ids.push(id.to_string());
}
}
}
}
ids.sort();
ids.iter()
.filter_map(|id| self.get_snapshot(id, false).ok().flatten())
.collect()
}
}
fn read_snapshot_file(
path: &str,
id: &str,
journal_bytes: i64,
) -> Option<(Map<String, Value>, i64)> {
let text = safe_read(path)?;
let parsed: Value = serde_json::from_str(&text).ok()?;
let mut obj = parsed.as_object()?.clone();
if obj.get(META_JOURNAL_BYTES).and_then(|v| v.as_i64()) != Some(journal_bytes) {
return None;
}
let next_seq = obj.get(META_NEXT_SEQ).and_then(|v| v.as_i64())?;
obj.remove(META_JOURNAL_BYTES);
obj.remove(META_NEXT_SEQ);
if obj.get("id").and_then(|v| v.as_str()) != Some(id) {
return None;
}
let mut merged = base_snapshot(id);
for (k, v) in obj {
merged.insert(k, v);
}
Some((merged, next_seq))
}
fn normalize_event(event: &Value, fallback_id: Option<&str>) -> Result<Map<String, Value>, String> {
let Some(obj) = event.as_object() else {
return Err("event object required".to_string());
};
let id: Option<String> = match obj.get("id") {
Some(Value::String(s)) if !s.is_empty() => Some(s.clone()),
Some(v) if !matches!(v, Value::Null | Value::String(_)) && truthy(v) => {
// JS: `event.id || fallbackId` then `typeof id !== 'string'` throws
let _ = v;
return Err("event id required".to_string());
}
_ => fallback_id.map(String::from),
};
let Some(id) = id else {
return Err("event id required".to_string());
};
match obj.get("type") {
Some(Value::String(t)) if !t.is_empty() => {}
_ => return Err("event type required".to_string()),
}
let mut out = obj.clone();
out.insert("id".to_string(), Value::String(id));
Ok(out)
}
fn truthy(v: &Value) -> bool {
match v {
Value::Null => false,
Value::Bool(b) => *b,
Value::Number(n) => n.as_f64().map(|f| f != 0.0).unwrap_or(false),
Value::String(s) => !s.is_empty(),
_ => true,
}
}
/// JS: baseSnapshot(id)
pub fn base_snapshot(id: &str) -> Map<String, Value> {
let v = json!({
"id": id,
"phase": "new",
"pageUrl": null,
"sourceFile": null,
"previewFile": null,
"previewMode": null,
"expectedVariants": 0,
"arrivedVariants": 0,
"visibleVariant": null,
"paramValues": {},
"pendingEventSeq": null,
"pendingEvent": null,
"deliveryLease": null,
"checkpointRevision": 0,
"browserCheckpointRevision": 0,
"publicationCheckpointRevision": 0,
"activeOwner": null,
"sourceMarkers": {},
"fallbackMode": null,
"generationPhase": null,
"generationCompletedAt": null,
"generationTimings": {},
"variantPlan": null,
"generationCanceled": false,
"generationCanceledAt": null,
"cancelReason": null,
"annotationArtifacts": [],
"mountedVariants": [],
"mountFailures": [],
"renderState": null,
"diagnostics": [],
"updatedAt": null,
});
v.as_object().cloned().unwrap_or_default()
}
fn derive_render_state(s: &Map<String, Value>) -> Value {
let mounted = s
.get("mountedVariants")
.and_then(|v| v.as_array())
.map(|a| a.len())
.unwrap_or(0);
let failures = s
.get("mountFailures")
.and_then(|v| v.as_array())
.map(|a| a.len())
.unwrap_or(0);
if mounted > 0 {
return json!("mounted");
}
if failures > 0 {
return json!("failed");
}
if truthy(s.get("generationCompletedAt").unwrap_or(&Value::Null)) {
return json!("pending");
}
Value::Null
}
/// JS: rebuildSnapshotFromJournal(journalPath, id) -> (snapshot, nextSeq)
pub fn rebuild_snapshot_from_journal(journal_path: &str, id: &str) -> (Map<String, Value>, i64) {
let mut snapshot = base_snapshot(id);
let mut diagnostics: Vec<Value> = Vec::new();
let mut next_seq: i64 = 1;
let Some(text) = safe_read(journal_path) else {
return (snapshot, next_seq);
};
for (i, line) in text.split('\n').enumerate() {
if impeccable_context::util::js_trim(line).is_empty() {
continue;
}
match serde_json::from_str::<Value>(line) {
Ok(entry) if entry.is_object() => {
if let Some(seq) = entry.get("seq").and_then(|s| s.as_i64()) {
next_seq = next_seq.max(seq + 1);
}
snapshot = apply_event(&snapshot, &entry);
}
Ok(_) => diagnostics.push(json!({ "error": "journal_parse_failed", "line": i + 1, "message": "entry is not object" })),
Err(e) => diagnostics.push(json!({ "error": "journal_parse_failed", "line": i + 1, "message": e.to_string() })),
}
}
if !diagnostics.is_empty() {
let mut d = snapshot
.get("diagnostics")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default();
d.extend(diagnostics);
snapshot.insert("diagnostics".to_string(), Value::Array(d));
}
(snapshot, next_seq)
}
fn is_nullish(v: Option<&Value>) -> bool {
matches!(v, None | Some(Value::Null))
}
/// `a ?? b`
fn coalesce<'a>(a: Option<&'a Value>, b: &'a Value) -> Value {
match a {
Some(v) if !v.is_null() => v.clone(),
_ => b.clone(),
}
}
fn get<'a>(m: &'a Map<String, Value>, k: &str) -> Option<&'a Value> {
m.get(k)
}
fn ts_ms(entry: &Value) -> Option<i64> {
// JS: Date.parse(entry.ts || '') || null
let ts = entry.get("ts").and_then(|t| t.as_str()).unwrap_or("");
date_parse_ms(ts).filter(|v| *v != 0)
}
fn to_pending_event(event: &Map<String, Value>) -> Value {
let mut p = event.clone();
p.shift_remove("token");
Value::Object(p)
}
fn push_diag(next: &mut Map<String, Value>, d: Value) {
let mut arr = next
.get("diagnostics")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default();
arr.push(d);
next.insert("diagnostics".to_string(), Value::Array(arr));
}
fn drop_diag(next: &mut Map<String, Value>, error: &str) {
let mut arr = next
.get("diagnostics")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default();
arr.retain(|d| d.get("error").and_then(|e| e.as_str()) != Some(error));
next.insert("diagnostics".to_string(), Value::Array(arr));
}
/// JS: applyEvent(snapshot, entry)
pub fn apply_event(snapshot: &Map<String, Value>, entry: &Value) -> Map<String, Value> {
let event: Map<String, Value> = match entry.get("event") {
Some(Value::Object(o)) => o.clone(),
Some(_) | None => entry.as_object().cloned().unwrap_or_default(),
};
let mut next = snapshot.clone();
// Spread copies (JS: fresh objects for the mutable containers).
for k in ["paramValues", "sourceMarkers", "generationTimings"] {
let v = next
.get(k)
.and_then(|v| v.as_object())
.cloned()
.unwrap_or_default();
next.insert(k.to_string(), Value::Object(v));
}
let vp = next.get("variantPlan").cloned().unwrap_or(Value::Null);
next.insert(
"variantPlan".to_string(),
if truthy(&vp) { vp } else { Value::Null },
);
for k in [
"annotationArtifacts",
"mountedVariants",
"mountFailures",
"diagnostics",
] {
let v = next
.get(k)
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default();
next.insert(k.to_string(), Value::Array(v));
}
let rs = next.get("renderState").cloned().unwrap_or(Value::Null);
next.insert("renderState".to_string(), rs);
next.insert(
"updatedAt".to_string(),
coalesce(
entry.get("ts").filter(|t| truthy(t)),
&Value::String(iso_now()),
),
);
let ev = |k: &str| event.get(k);
let evt_type = event
.get("type")
.and_then(|t| t.as_str())
.unwrap_or("")
.to_string();
let phase_now = next
.get("phase")
.and_then(|p| p.as_str())
.unwrap_or("")
.to_string();
let canceled = truthy(next.get("generationCanceled").unwrap_or(&Value::Null));
let fenced = GENERATION_FENCED_SESSION_PHASES.contains(&phase_now.as_str());
let seq = entry.get("seq").cloned();
let now_or_ts = |entry: &Value| -> Value {
match ts_ms(entry) {
Some(ms) => json!(ms),
None => json!(now_ms() as i64),
}
};
macro_rules! set_if {
($k:expr, $v:expr) => {{
let cur = next.get($k).cloned().unwrap_or(Value::Null);
next.insert($k.to_string(), coalesce($v, &cur));
}};
}
macro_rules! set {
($k:expr, $v:expr) => {
next.insert($k.to_string(), $v);
};
}
match evt_type.as_str() {
"generate" => {
set!("phase", json!("generate_requested"));
// `origin: "agent"` marks a Go the generate verb fired; the
// accept pipeline bakes those sessions itself. A plain Go
// carries no origin and its snapshot stays exactly as it was.
if let Some(origin) = ev("origin").filter(|v| truthy(v)) {
set!("origin", origin.clone());
}
set_if!("pageUrl", ev("pageUrl"));
set_if!("expectedVariants", ev("count"));
set_if!("pendingEventSeq", seq.as_ref());
set!("pendingEvent", to_pending_event(&event));
set!("variantPlan", Value::Null);
set!("mountedVariants", json!([]));
set!("mountFailures", json!([]));
set!("renderState", Value::Null);
if let Some(sp) = ev("screenshotPath").filter(|v| truthy(v)) {
let mut arts = next
.get("annotationArtifacts")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default();
let exists_already = arts.iter().any(|a| {
a.get("path") == Some(sp)
&& a.get("type").and_then(|t| t.as_str()) == Some("screenshot")
});
if !exists_already {
arts.push(json!({ "type": "screenshot", "path": sp }));
}
set!("annotationArtifacts", Value::Array(arts));
}
}
"variant_plan" => {
if !canceled && !fenced {
set_if!("variantPlan", ev("plan"));
}
}
"detector_waivers" => {
if !canceled && !fenced {
let mut w = next
.get("detectorWaivers")
.and_then(|v| v.as_array())
.cloned()
.unwrap_or_default();
if let Some(arr) = ev("waivers").and_then(|v| v.as_array()) {
w.extend(arr.iter().cloned());
}
set!("detectorWaivers", Value::Array(w));
}
}
"agent_phase" => {
set_if!("generationPhase", ev("phase"));
if let Some(phase) = ev("phase").filter(|v| truthy(v)) {
let key = match phase {
Value::String(s) => s.clone(),
other => other.to_string(),
};
let at = match ev("at") {
Some(v) if !v.is_null() => v.clone(),
_ => ts_ms(entry).map(|ms| json!(ms)).unwrap_or(Value::Null),
};
let dur = coalesce(ev("durationMs"), &Value::Null);
let mut timings = next
.get("generationTimings")
.and_then(|v| v.as_object())
.cloned()
.unwrap_or_default();
timings.insert(key, json!({ "at": at, "durationMs": dur }));
set!("generationTimings", Value::Object(timings));
}
}
"variants_ready" | "agent_done" => {
let carbonize = ev("carbonize") == Some(&Value::Bool(true));
if (canceled || fenced)
&& !(evt_type == "agent_done" && carbonize && phase_now == "accept_requested")
{
push_diag(
&mut next,
json!({ "error": "late_generation_event_ignored", "type": evt_type, "phase": phase_now }),
);
} else {
set!(
"phase",
json!(if carbonize {
"carbonize_required"
} else {
"variants_ready"
})
);
let at = match ev("at") {
Some(v) if !v.is_null() => v.clone(),
_ => now_or_ts(entry),
};
set!("generationCompletedAt", at);
let sf = coalesce(
ev("sourceFile"),
&coalesce(ev("file"), next.get("sourceFile").unwrap_or(&Value::Null)),
);
set!("sourceFile", sf);
set_if!("previewFile", ev("previewFile"));
set_if!("previewMode", ev("previewMode"));
let arrived = match ev("arrivedVariants") {
Some(v) if !v.is_null() => v.clone(),
_ => {
let expected = next.get("expectedVariants").cloned().unwrap_or(Value::Null);
if truthy(&expected) {
expected
} else {
let a = next.get("arrivedVariants").cloned().unwrap_or(Value::Null);
if truthy(&a) {
a
} else {
json!(0)
}
}
}
};
set!("arrivedVariants", arrived);
set!("pendingEventSeq", Value::Null);
set!("pendingEvent", Value::Null);
if carbonize {
let file = match ev("file") {
Some(v) if truthy(v) => v.clone(),
_ => Value::Null,
};
push_diag(
&mut next,
json!({
"error": "carbonize_cleanup_required",
"file": file,
"message": "Accepted variant still has carbonize markers that must be folded into source CSS.",
}),
);
}
let rs = derive_render_state(&next);
set!("renderState", rs);
}
}
"variant_mounted" => {
let variant = js_number(ev("variant"));
match variant {
Some(v) if v.fract() == 0.0 && v >= 1.0 => {
let v = v as i64;
let mut mounted: Vec<i64> = next
.get("mountedVariants")
.and_then(|m| m.as_array())
.map(|a| a.iter().filter_map(|x| x.as_i64()).collect())
.unwrap_or_default();
if !mounted.contains(&v) {
mounted.push(v);
mounted.sort();
}
set!("mountedVariants", json!(mounted));
let rs = derive_render_state(&next);
set!("renderState", rs);
}
_ => {
push_diag(
&mut next,
json!({ "error": "malformed_mount_ack", "type": evt_type, "variant": coalesce(ev("variant"), &Value::Null) }),
);
}
}
}
"variant_mount_failed" => {
let variant = js_number(ev("variant"));
match variant {
Some(v) if v.fract() == 0.0 && v >= 1.0 => {
let mut failures = next
.get("mountFailures")
.and_then(|m| m.as_array())
.cloned()
.unwrap_or_default();
let at = match ev("at") {
Some(a) if !a.is_null() => a.clone(),
_ => now_or_ts(entry),
};
failures.push(json!({
"variant": v as i64,
"url": ev("url").and_then(|u| u.as_str()).map(|s| json!(s)).unwrap_or(Value::Null),
"error": ev("error").and_then(|u| u.as_str()).map(|s| json!(s)).unwrap_or(Value::Null),
"at": at,
}));
if failures.len() > MOUNT_FAILURE_HISTORY {
failures = failures[failures.len() - MOUNT_FAILURE_HISTORY..].to_vec();
}
set!("mountFailures", Value::Array(failures));
let rs = derive_render_state(&next);
set!("renderState", rs);
if !truthy(next.get("pendingEvent").unwrap_or(&Value::Null)) {
set!("pendingEvent", to_pending_event(&event));
}
}
_ => {
push_diag(
&mut next,
json!({ "error": "malformed_mount_ack", "type": evt_type, "variant": coalesce(ev("variant"), &Value::Null) }),
);
}
}
}
"checkpoint" => {
if canceled || fenced {
push_diag(
&mut next,
json!({ "error": "checkpoint_after_terminal_ignored", "phase": coalesce(ev("phase"), &Value::Null), "revision": coalesce(ev("revision"), &Value::Null) }),
);
} else {
let publication = ev("revisionDomain").and_then(|v| v.as_str())
== Some("publication")
|| (ev("reason").and_then(|v| v.as_str()) == Some("variants_progress")
&& !truthy(ev("owner").unwrap_or(&Value::Null)));
let domain = if publication {
"publication"
} else {
"browser"
};
let field = if publication {
"publicationCheckpointRevision"
} else {
"browserCheckpointRevision"
};
let current_revision: f64 = match next.get(field) {
Some(v) if !v.is_null() => js_number(Some(v)).unwrap_or(f64::NAN),
_ => {
if domain == "browser" {
match next.get("checkpointRevision") {
Some(v) if !v.is_null() => js_number(Some(v)).unwrap_or(f64::NAN),
_ => 0.0,
}
} else {
0.0
}
}
};
let rev_val = coalesce(ev("revision"), &json!(0));
let rev_num = js_number(Some(&rev_val)).unwrap_or(f64::NAN);
if rev_num >= current_revision {
set_if!("phase", ev("phase"));
let cur = next.get(field).cloned().unwrap_or(Value::Null);
let stored = coalesce(
ev("revision"),
&if current_revision.is_nan() {
cur
} else {
crate::util::js_num(current_revision)
},
);
set!(field, stored);
if domain == "browser" {
set_if!("checkpointRevision", ev("revision"));
set_if!("activeOwner", ev("owner"));
}
set_if!("arrivedVariants", ev("arrivedVariants"));
if domain == "browser" {
set_if!("visibleVariant", ev("visibleVariant"));
}
set_if!("sourceFile", ev("sourceFile"));
set_if!("previewFile", ev("previewFile"));
set_if!("previewMode", ev("previewMode"));
if domain == "browser" {
if let Some(pv) = ev("paramValues").filter(|v| truthy(v)) {
let copy = pv
.as_object()
.cloned()
.map(Value::Object)
.unwrap_or_else(|| json!({}));
set!("paramValues", copy);
}
}
} else {
push_diag(
&mut next,
json!({ "error": "stale_checkpoint_ignored", "revision": ev("revision").cloned().unwrap_or(Value::Null), "revisionDomain": domain }),
);
// JS: `revision: event.revision` (undefined is dropped by JSON)
if is_nullish(ev("revision")) && ev("revision").is_none() {
if let Some(Value::Array(arr)) = next.get_mut("diagnostics") {
if let Some(Value::Object(last)) = arr.last_mut() {
last.shift_remove("revision");
}
}
}
}
}
}
"accept" | "accept_intent" => {
set!("phase", json!("accept_requested"));
set!("generationCanceled", json!(true));
let at = match ev("at") {
Some(a) if !a.is_null() => a.clone(),
_ => now_or_ts(entry),
};
set!("generationCanceledAt", at);
set!("cancelReason", json!("accept"));
let vid = coalesce(
ev("variantId"),
next.get("visibleVariant").unwrap_or(&Value::Null),
);
let n = js_number(Some(&vid));
set!(
"visibleVariant",
n.map(crate::util::js_num).unwrap_or(Value::Null)
);
if let Some(pv) = ev("paramValues").filter(|v| truthy(v)) {
let copy = pv
.as_object()
.cloned()
.map(Value::Object)
.unwrap_or_else(|| json!({}));
set!("paramValues", copy);
}
set_if!("pendingEventSeq", seq.as_ref());
set!("pendingEvent", to_pending_event(&event));
}
"manual_edit_apply" => {
set!("phase", json!("manual_edit_apply_requested"));
set_if!("pageUrl", ev("pageUrl"));
set_if!("pendingEventSeq", seq.as_ref());
set!("pendingEvent", to_pending_event(&event));
}
"steer" => {
set!("phase", json!("steer_requested"));
set_if!("pageUrl", ev("pageUrl"));
set_if!("pendingEventSeq", seq.as_ref());
set!("pendingEvent", to_pending_event(&event));
}
"carbonize_cleanup" => {
set!("phase", json!("carbonize_cleanup_requested"));
set_if!("sourceFile", ev("file"));
set_if!("pendingEventSeq", seq.as_ref());
set!("pendingEvent", to_pending_event(&event));
}
"steer_done" => {
set!("phase", json!("steer_done"));
let sf = coalesce(
ev("sourceFile"),
&coalesce(ev("file"), next.get("sourceFile").unwrap_or(&Value::Null)),
);
set!("sourceFile", sf);
set_if!("previewFile", ev("previewFile"));
set_if!("previewMode", ev("previewMode"));
// JS: next.message = event.message ?? next.message (key appears
// even when undefined... no: assigning undefined creates the key,
// JSON.stringify drops it).
match ev("message") {
Some(v) if !v.is_null() => {
set!("message", v.clone());
}
_ => {
if let Some(cur) = next.get("message").cloned() {
set!("message", cur);
}
}
}
set!("pendingEventSeq", Value::Null);
set!("pendingEvent", Value::Null);
}
"discard" => {
set!("phase", json!("discard_requested"));
set!("generationCanceled", json!(true));
let at = match ev("at") {
Some(a) if !a.is_null() => a.clone(),
_ => now_or_ts(entry),
};
set!("generationCanceledAt", at);
set!("cancelReason", json!("discard"));
set_if!("pendingEventSeq", seq.as_ref());
set!("pendingEvent", to_pending_event(&event));
}
"discarded" => {
set!("phase", json!("discarded"));
set!("pendingEventSeq", Value::Null);
set!("pendingEvent", Value::Null);
drop_diag(&mut next, "carbonize_cleanup_required");
}
"complete" => {
set!("phase", json!("completed"));
let sf = coalesce(
ev("sourceFile"),
&coalesce(ev("file"), next.get("sourceFile").unwrap_or(&Value::Null)),
);
set!("sourceFile", sf);
set_if!("previewFile", ev("previewFile"));
set_if!("previewMode", ev("previewMode"));
set!("pendingEventSeq", Value::Null);
set!("pendingEvent", Value::Null);
drop_diag(&mut next, "carbonize_cleanup_required");
}
"agent_error" => {
if canceled && ev("sourceEventType").and_then(|v| v.as_str()) == Some("generate") {
push_diag(
&mut next,
json!({ "error": "late_generation_event_ignored", "type": evt_type, "phase": phase_now }),
);
} else {
set!("phase", json!("agent_error"));
set!("pendingEventSeq", Value::Null);
set!("pendingEvent", Value::Null);
let msg = match ev("message") {
Some(v) if truthy(v) => v.clone(),
_ => json!("unknown agent error"),
};
push_diag(&mut next, json!({ "error": "agent_error", "message": msg }));
}
}
_ => {
push_diag(
&mut next,
json!({ "error": "unknown_event_type", "type": ev("type").cloned().unwrap_or(Value::Null) }),
);
if ev("type").is_none() {
if let Some(Value::Array(arr)) = next.get_mut("diagnostics") {
if let Some(Value::Object(last)) = arr.last_mut() {
last.shift_remove("type");
}
}
}
}
}
next
}
fn write_snapshot(path: &str, snapshot: &Map<String, Value>, journal_bytes: i64, next_seq: i64) {
let mut payload = snapshot.clone();
payload.insert(META_JOURNAL_BYTES.to_string(), json!(journal_bytes));
payload.insert(META_NEXT_SEQ.to_string(), json!(next_seq));
let _ = crate::util::write_file(
path,
&format!("{}\n", crate::util::json_pretty(&Value::Object(payload))),
);
}
/// The pending event's `id`/`type` as strings (helper for status/resume).
pub fn get_str<'a>(m: &'a Map<String, Value>, k: &str) -> Option<&'a str> {
get(m, k).and_then(|v| v.as_str())
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn journal_entry(seq: i64, event: Value) -> Value {
json!({ "seq": seq, "ts": "2026-01-01T00:00:00.000Z", "event": event })
}
fn has_diag(snapshot: &Map<String, Value>, error: &str) -> bool {
snapshot
.get("diagnostics")
.and_then(|v| v.as_array())
.map(|a| {
a.iter()
.any(|d| d.get("error").and_then(|e| e.as_str()) == Some(error))
})
.unwrap_or(false)
}
fn replay(id: &str, events: &[Value]) -> Map<String, Value> {
let mut snap = base_snapshot(id);
for entry in events {
snap = apply_event(&snap, entry);
}
snap
}
fn accept_carbonize_done(id: &str, terminal: &str) -> Map<String, Value> {
replay(
id,
&[
journal_entry(
1,
json!({ "id": id, "type": "accept", "variantId": 2 }),
),
journal_entry(
2,
json!({ "id": id, "type": "agent_done", "carbonize": true, "file": "index.html" }),
),
journal_entry(3, json!({ "id": id, "type": terminal })),
],
)
}
#[test]
fn complete_drops_carbonize_cleanup_required() {
let snap = accept_carbonize_done("ab12cd34", "complete");
assert_eq!(snap.get("phase").and_then(|p| p.as_str()), Some("completed"));
assert!(!has_diag(&snap, "carbonize_cleanup_required"));
}
#[test]
fn discarded_drops_carbonize_cleanup_required() {
let snap = accept_carbonize_done("ab12cd34", "discarded");
assert_eq!(snap.get("phase").and_then(|p| p.as_str()), Some("discarded"));
assert!(!has_diag(&snap, "carbonize_cleanup_required"));
}
}
Reference in New Issue View Git Blame Copy Permalink