diff --git a/lib/components/fabro-petri/src/projection.rs b/lib/components/fabro-petri/src/projection.rs deleted file mode 100644 index 54183cc9f..000000000 --- a/lib/components/fabro-petri/src/projection.rs +++ /dev/null @@ -1,1834 +0,0 @@ -//! The projection of a Petri run: Petri's public events and Fabro's platform -//! records folded into the view Fabro's read side serves. -//! -//! The fold is pure. [`RunView`] holds the [`RunProjection`] the API serves -//! (`GET /runs/{id}/state`, the run list through its summary) and the -//! bookkeeping the fold needs between items ([`FoldState`]): which Petri -//! firing each stage is, which invocation each execution belongs to and -//! whether it is a parallel branch, which stage asked each open question. -//! Both halves are stored by the projector and reloaded for the next pass, -//! so a pass folds only the items past the committed positions. -//! -//! The mapping follows `VIEWS.md`, row by row. The stage key is `(execution, -//! firing)`; Fabro's `StageId` (`node@visit`) is the display label the -//! `RunProjection` keys stages by, and a label two firings would share (two -//! child invocations with the same node name and visit) is made unique by -//! naming the execution. What the matrix leaves default is left default -//! here and named in the crate's README. -//! -//! Every item the fold sees carries the delivery sequence the projector -//! assigned it (`stream_seq`), which a checkpoint keeps as its `seq`. A -//! stage's `first_event_seq`, the key the stage list sorts by, is not the -//! delivery sequence: two logs' records can be committed in an order that -//! differs from their recording times by a few positions, and the view -//! built live must equal the view rebuilt from the records alone. It is the -//! milliseconds from the run's creation to the stage's `visit.started`, -//! plus one, which is the same however the records were delivered. - -use std::collections::{BTreeMap, BTreeSet}; - -use chrono::{DateTime, TimeZone as _, Utc}; -use fabro_store::platform_records::{ - PlatformRecord, RunLifecycleKind, RunLifecycleRecord, StoredPlatformRecord, -}; -use fabro_types::settings::run::RunEnvironmentSettings; -use fabro_types::{ - BlockedReason, CheckpointRecord as ViewCheckpoint, CodingAgentEvent, CodingEvent, Conclusion, - FailureCategory, FailureDetail, FailureReason, InterviewOption, InterviewQuestionRecord, - ModelRef, ModelUsage, ParallelBranchId, ParallelBranchResult, PendingInterviewRecord, - PullRequestCreation, PullRequestCreationStatus, PullRequestLink, ReviewTarget, - ReviewTargetKind, RunApproval, RunApprovalState, RunArtifact, RunControlAction, RunDiff, - RunFailure, RunId, RunProjection, RunSandbox, RunSandboxFailure, RunSandboxInstance, - RunSandboxPlan, RunSandboxRuntime, RunStatus, RunTiming, SandboxProviderKind, StageCompletion, - StageHandler, StageId, StageInferenceProjection, StageModelUsage, StageOutcome, - StageProjection, StageState, StageTiming, StartRecord, SuccessReason, ToolCategory, ToolSource, - ToolSummary, first_event_seq, format_blob_ref, parse_blob_ref, timing, usage_rollup, -}; -use lithos_llm::catalog::{ModelId, ProviderId}; -use lithos_llm::types::Usage; -use petri_execution::events::{Derived, NodeRef, Parsed, RunEvent, Subject, ViewEvent, WaitState}; -use petri_execution::{CoordinatorEvent, ExecutionId, InvocationId}; -use petri_runtime::engine::{Admission, Event}; -use petri_runtime::ir::{Metrics, SandboxInstance, Status, StepEvent}; -use petri_runtime::steps::QuestionReference; -use serde::{Deserialize, Serialize}; -use serde_json::Value; -use tracing::debug; - -use crate::interview::question_type; - -/// One item the projector hands the fold, with its delivery sequence. -pub enum Item<'a> { - Petri(&'a RunEvent), - Platform(&'a StoredPlatformRecord), -} - -/// A stage as the fold knows it: its label in the projection, and what it -/// learned about it. -#[derive(Clone, Debug, Serialize, Deserialize)] -pub struct StageRef { - pub stage_id: StageId, - /// Whether the stage is a logical one the projection shows, or a - /// lowering node it keeps off the list. - pub shown: bool, - /// The node's instance name and visit, for the collision rule. - pub node_name: String, - pub visit: u32, -} - -/// What the fold knows about one invocation. -#[derive(Clone, Debug, Default, Serialize, Deserialize)] -pub struct InvocationRef { - /// The calling execution and firing, for a nested invocation. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub parent: Option<(u64, u64)>, - /// The parallel group and branch index, for a branch child. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub branch: Option<(StageId, u32)>, - /// The result the invocation recorded, for the root. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub failure: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub output: Option, -} - -/// Whether the run's durable record is whole, as the projector last read it. -#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] -pub struct RecordHealth { - pub complete: bool, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub incomplete: Vec, -} - -/// The fold's bookkeeping between items. -#[derive(Clone, Debug, Default, Serialize, Deserialize)] -pub struct FoldState { - /// Stages by `":"`. - #[serde(default)] - pub stages: BTreeMap, - /// Labels taken, so a second firing with the same name and visit gets - /// its own. - #[serde(default)] - pub labels: BTreeSet, - #[serde(default)] - pub invocations: BTreeMap, - /// Which invocation each execution belongs to. - #[serde(default)] - pub executions: BTreeMap, - /// Open questions by id: the stage that asked. - #[serde(default)] - pub questions: BTreeMap, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub root: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option, - /// The run's recorded finish, when Petri recorded one. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished: Option, - /// The run branch and base sha, when they arrive before `run.started`. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub run_branch: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub base_sha: Option, - #[serde(default)] - pub checkpoints: u32, - /// The run's diff as its `run.diff` record gave it, whichever side of - /// the run's finish it arrived on. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub run_diff: Option, - #[serde(default)] - pub health: RecordHealth, - /// Firings (`":"`) whose attempt has recorded a - /// finish: what a position-keyed platform record may be streamed - /// behind. - #[serde(default)] - pub finished_firings: BTreeSet, - /// Whether the run's sandbox still exists after its release - /// (`scope.released` `retained`): kept stopped, or deleted. Absent until - /// the root invocation's lease was released. The view carries the same - /// fact as `RunSandboxInstance.retained`. - #[serde(default, skip_serializing_if = "Option::is_none")] - pub sandbox_retained: Option, -} - -impl FoldState { - /// Whether Petri recorded the run's finish. - #[must_use] - pub fn finished_run(&self) -> bool { - self.finished.is_some() - } -} - -/// The view of one run: what the API serves and what the fold keeps. -#[derive(Clone, Debug)] -pub struct RunView { - pub projection: Option, - pub state: FoldState, -} - -impl RunView { - #[must_use] - pub fn new() -> Self { - Self { - projection: None, - state: FoldState::default(), - } - } - - /// Fold one item at its delivery sequence. - pub fn fold(&mut self, item: &Item<'_>, stream_seq: u64) { - match item { - Item::Platform(record) => self.fold_platform(record, stream_seq), - Item::Petri(event) => self.fold_petri(event), - } - } - - /// The run's projection, once its `run.created` record was folded. - #[must_use] - pub fn projection(&self) -> Option<&RunProjection> { - self.projection.as_ref() - } - - // ── Platform records ──────────────────────────────────────────────── - - fn fold_platform(&mut self, stored: &StoredPlatformRecord, stream_seq: u64) { - let at = millis(stored.recorded_at); - if let PlatformRecord::RunCreated(created) = &stored.record { - let title = created - .title - .clone() - .unwrap_or_else(|| fabro_types::infer_run_title(created.spec.graph.goal())); - let mut projection = RunProjection::new(title, created.spec.clone(), at); - projection.parent_id = created.parent_id; - projection.retried_from = created.retried_from; - projection.web_url.clone_from(&created.web_url); - projection.sandbox = Some(RunSandbox::planned(sandbox_plan( - &projection.spec.settings.run.environment, - ))); - self.projection = Some(projection); - return; - } - let Some(projection) = self.projection.as_mut() else { - debug!( - seq = stored.seq, - kind = %stored.record.kind(), - "platform record before run.created; not folded" - ); - return; - }; - touch(projection, at); - match &stored.record { - PlatformRecord::RunLifecycle(record) => fold_lifecycle(projection, record, at), - PlatformRecord::RunTitle(record) => projection.title.clone_from(&record.title), - PlatformRecord::RunParent(record) => projection.parent_id = record.parent_id, - PlatformRecord::RunArchived => projection.archived_at = Some(at), - PlatformRecord::RunUnarchived => projection.archived_at = None, - PlatformRecord::RunSuperseded(record) => { - projection.superseded_by = Some(record.new_run_id); - } - PlatformRecord::RunCreated(_) - | PlatformRecord::RunNotice(_) - | PlatformRecord::InterviewAnswered(_) - | PlatformRecord::NotificationSent(_) - | PlatformRecord::RunPaired(_) => {} - PlatformRecord::RunBranch(record) => { - self.state.run_branch.clone_from(&record.run_branch); - self.state.base_sha.clone_from(&record.base_sha); - if let Some(start) = projection.start.as_mut() { - start.run_branch.clone_from(&record.run_branch); - start.base_sha.clone_from(&record.base_sha); - } - } - PlatformRecord::GitIdentity(record) => { - projection.git_identity = Some(record.identity.clone()); - } - PlatformRecord::Checkpoint(record) => { - self.state.checkpoints = self.state.checkpoints.saturating_add(1); - let stage = self - .state - .stages - .get(&stage_key(record.execution, record.firing)); - let current_node = stage.map_or_else(String::new, |stage| stage.node_name.clone()); - let stage_id = stage - .filter(|stage| stage.shown) - .map(|stage| stage.stage_id.clone()); - let checkpoint = fabro_types::Checkpoint { - timestamp: at, - current_node: current_node.clone(), - git_commit_sha: record.git_commit_sha.clone(), - }; - // The patch stays in the blob table; the view carries its - // reference for a reader to resolve. - let patch = record.patch_blob.as_ref().map(format_blob_ref); - if let Some(stage) = stage_id.and_then(|stage_id| projection.stage_mut(&stage_id)) { - if patch.is_some() { - stage.diff.clone_from(&patch); - } - } - projection.checkpoints.push(ViewCheckpoint { - seq: u32::try_from(stream_seq).unwrap_or(u32::MAX), - checkpoint, - diff: RunDiff { - patch, - summary: record.diff_summary, - }, - }); - } - PlatformRecord::ArtifactCollected(record) => { - let stage = self - .state - .stages - .get(&stage_key(record.execution, record.firing)); - let Some(stage_id) = stage.map(|stage| stage.stage_id.clone()) else { - debug!( - seq = stored.seq, - path = record.path, - "artifact record for an unknown firing; not folded" - ); - return; - }; - projection.artifacts.push(RunArtifact { - stage_id, - retry: record.attempt, - relative_path: record.path.clone(), - size: record.bytes, - blob: record.blob, - }); - } - PlatformRecord::RunDiff(record) => { - let diff = RunDiff { - patch: record.patch_blob.as_ref().map(format_blob_ref), - summary: record.diff_summary, - }; - if let Some(conclusion) = projection.conclusion.as_mut() { - conclusion.diff = diff.clone(); - } - self.state.run_diff = Some(diff); - } - PlatformRecord::PullRequestRequested(record) => { - projection.pull_request_creation = Some(PullRequestCreation { - id: record.creation_id, - status: PullRequestCreationStatus::Pending, - model: record.model.clone(), - force: record.force, - requested_at: at, - updated_at: at, - pull_request: None, - error: None, - }); - } - PlatformRecord::PullRequestCreated(record) => { - let link = PullRequestLink { - owner: record.owner.clone(), - repo: record.repo.clone(), - number: record.number, - }; - projection.pull_request = Some(link.clone()); - if let Some(creation) = projection - .pull_request_creation - .as_mut() - .filter(|creation| creation.is_pending()) - { - creation.succeed(link, at); - } - } - PlatformRecord::PullRequestFailed(record) => { - if let Some(creation) = - projection - .pull_request_creation - .as_mut() - .filter(|creation| { - creation.is_pending() - && record - .creation_id - .is_none_or(|creation_id| creation_id == creation.id) - }) - { - creation.fail(record.error.clone(), at); - } - } - PlatformRecord::PullRequestLinked(record) => { - let link = record.link(); - projection.pull_request = Some(link.clone()); - if let Some(creation) = projection - .pull_request_creation - .as_mut() - .filter(|creation| creation.is_pending()) - { - creation.succeed(link, at); - } - } - PlatformRecord::PullRequestUnlinked(_) => { - projection.pull_request = None; - projection.pull_request_creation = None; - } - } - } - - // ── Petri events ──────────────────────────────────────────────────── - - fn fold_petri(&mut self, event: &RunEvent) { - let at = millis(event.recorded_at); - if let Some(record) = event.coordinator() { - self.fold_coordinator(record, event, at); - } else if let Some(engine) = event.engine() { - self.fold_engine(engine, event, at); - } else if let Some(view) = event.view() { - self.fold_view(view, event, at); - } - if let Some(projection) = self.projection.as_mut() { - touch(projection, at); - } - } - - fn fold_coordinator(&mut self, record: &CoordinatorEvent, event: &RunEvent, at: DateTime) { - match record { - CoordinatorEvent::RunStarted { - root, forked_from, .. - } => { - self.state.root = Some(root.raw()); - self.state.started_at = Some(event.recorded_at); - if let Some(projection) = self.projection.as_mut() { - // A fork's declaration names its source; a parse failure - // means the source was not a Fabro run, which the - // projection cannot show. - projection.forked_from = forked_from.as_ref().and_then(|origin| { - Some(fabro_types::ForkOrigin { - source_run_id: origin.source.as_str().parse().ok()?, - execution: origin.position.execution.raw(), - firing: origin.position.firing.raw(), - rerun_last: origin.rerun_last, - }) - }); - apply_status(projection, RunStatus::Running, at); - projection.start = Some(StartRecord { - start_time: at, - run_branch: self.state.run_branch.clone(), - base_sha: self.state.base_sha.clone(), - }); - // The scope's sandbox is acquired next; `scope.acquired` - // or `scope.failed` settles it. - if let Some(sandbox) = projection.sandbox.take() { - projection.sandbox = Some(RunSandbox::initializing(sandbox.plan().clone())); - } - } - } - CoordinatorEvent::InvocationDeclared { invocation, .. } => { - let mut info = InvocationRef::default(); - if let Some(parent) = &event.context.parent { - info.parent = Some((parent.execution.raw(), parent.firing.raw())); - if let Some((fork_firing, index)) = branch_slot(&parent.slot) { - let group = self - .state - .stages - .get(&stage_key(parent.execution.raw(), fork_firing)) - .map(|stage| stage.stage_id.clone()); - if let Some(group) = group { - info.branch = Some((group, index)); - } - } - } - self.state.invocations.insert(invocation.raw(), info); - } - CoordinatorEvent::ExecutionDeclared { - execution, - invocation, - .. - } => { - self.state - .executions - .insert(execution.raw(), invocation.raw()); - } - CoordinatorEvent::InvocationFinished { invocation, result } => { - let info = self.state.invocations.entry(invocation.raw()).or_default(); - info.failure = result - .failure - .as_ref() - .map(|failure| failure.message.clone()); - info.output = Some(result.output.clone()); - } - CoordinatorEvent::RunPaused => { - if let Some(projection) = self.projection.as_mut() { - let prior_block = match projection.status { - RunStatus::Blocked { blocked_reason } => Some(blocked_reason), - _ => None, - }; - apply_status(projection, RunStatus::Paused { prior_block }, at); - if projection.pending_control == Some(RunControlAction::Pause) { - projection.pending_control = None; - } - } - } - CoordinatorEvent::RunUnpaused => { - if let Some(projection) = self.projection.as_mut() { - let next = match projection.status { - RunStatus::Paused { - prior_block: Some(blocked_reason), - } => RunStatus::Blocked { blocked_reason }, - _ => RunStatus::Running, - }; - apply_status(projection, next, at); - if projection.pending_control == Some(RunControlAction::Unpause) { - projection.pending_control = None; - } - } - } - CoordinatorEvent::RunFinished { status } => { - self.state.finished = Some(status.to_string()); - self.conclude(status.to_string().as_str(), at); - } - // ── Sandbox: the retention outcome (VIEWS.md "Sandbox") ───────── - // The instance stays on `Run.sandbox`: it names what ran, and - // `retained` says whether it still exists. - CoordinatorEvent::ScopeReleased { - invocation, - retained, - .. - } => { - if Some(invocation.raw()) == self.state.root { - self.state.sandbox_retained = Some(*retained); - if let Some(sandbox) = self - .projection - .as_mut() - .and_then(|projection| projection.sandbox.as_mut()) - { - sandbox.set_retained(*retained); - } - } - } - CoordinatorEvent::GraphRegistered { .. } - | CoordinatorEvent::ExecutionFinished { .. } - | CoordinatorEvent::InvocationCancelRequested { .. } - | CoordinatorEvent::RunNoteRecorded { .. } => {} - } - } - - /// The run's conclusion, from its recorded finish and what the stages - /// summed to. - fn conclude(&mut self, status: &str, at: DateTime) { - let Some(projection) = self.projection.as_mut() else { - return; - }; - let root = self - .state - .root - .and_then(|root| self.state.invocations.get(&root)); - let failure_message = root.and_then(|root| root.failure.clone()); - let (run_status, outcome, failure) = match status { - "success" => ( - RunStatus::Succeeded { - reason: SuccessReason::Completed, - }, - StageOutcome::Succeeded, - None, - ), - "cancelled" => ( - RunStatus::Failed { - reason: FailureReason::Cancelled, - }, - StageOutcome::Failed { - retry_requested: false, - }, - Some(RunFailure { - reason: FailureReason::Cancelled, - detail: FailureDetail::new( - failure_message - .clone() - .unwrap_or_else(|| "the run was cancelled".to_string()), - FailureCategory::Canceled, - ), - }), - ), - _ => ( - RunStatus::Failed { - reason: FailureReason::WorkflowError, - }, - StageOutcome::Failed { - retry_requested: false, - }, - Some(RunFailure { - reason: FailureReason::WorkflowError, - detail: FailureDetail::new( - failure_message - .clone() - .unwrap_or_else(|| "the run failed".to_string()), - FailureCategory::Deterministic, - ), - }), - ), - }; - apply_status(projection, run_status, at); - projection.pending_control = None; - projection.pending_interviews.clear(); - let rollup = usage_rollup::usage_rollup_from_projection(projection); - let (stages, total_retries) = rollup.conclusion_stages(projection); - let wall_time_ms = self.state.started_at.map_or(0, |started| { - u64::try_from(at.timestamp_millis()) - .unwrap_or(0) - .saturating_sub(started) - }); - let timing = RunTiming::new( - wall_time_ms, - rollup.timing.inference_time_ms, - rollup.timing.tool_time_ms, - ); - let last_checkpoint = projection.checkpoints.last(); - projection.conclusion = Some(Conclusion { - timestamp: at, - status: outcome, - timing, - failure, - final_git_commit_sha: last_checkpoint - .and_then(|checkpoint| checkpoint.checkpoint.git_commit_sha.clone()), - stages, - usage: rollup.usage_if_present(), - total_retries, - diff: self - .state - .run_diff - .clone() - .or_else(|| last_checkpoint.map(|checkpoint| checkpoint.diff.clone())) - .unwrap_or_default(), - }); - } - - fn fold_engine(&mut self, engine: &Event, event: &RunEvent, at: DateTime) { - let Some(execution) = event.context.execution else { - return; - }; - match engine { - Event::AdmissionDecided { decision, .. } => { - if let Admission::Skip { outcome } = decision { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - stage.state = StageState::Skipped; - stage.completion = Some(StageCompletion { - outcome: StageOutcome::Skipped, - notes: None, - failure_reason: failure_message(&outcome.status), - timestamp: at, - }); - } - } - } - Event::StepStarted { attempt, .. } => { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - if attempt.raw() > 1 { - stage.clear_live_timing(); - stage.output = None; - stage.output_bytes = None; - } - stage.state = StageState::Running; - stage.live_streaming = Some(true); - } - } - Event::StepProgressRecorded { ev, .. } => { - self.fold_progress(execution, event, ev, at); - } - Event::StepFinished { - firing, - attempt, - outcome, - } => { - self.state - .finished_firings - .insert(stage_key(execution.raw(), firing.raw())); - let is_final = matches!( - event.derived, - Some(Derived::StepFinished { is_final: true, .. }) - ); - let node_name = event - .subject - .as_ref() - .map(|subject| subject.node.name.to_string()); - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - // The step's output: a string, or a command's `stdout`, - // either of which is a `blob://` reference when the - // step offloaded it. The reference stays as it is; the - // bytes it names are the live log's. - let output = outcome - .output - .as_str() - .or_else(|| outcome.output.get("stdout").and_then(Value::as_str)); - if let Some(output) = output { - if parse_blob_ref(output).is_none() { - stage.output_bytes = Some(output.len() as u64); - } - stage.output = Some(output.to_string()); - } - // A simulated step (a dry run) answers with its text. - let simulated = outcome - .output - .get("simulated") - .and_then(Value::as_bool) - .unwrap_or(false); - if simulated - && matches!( - stage.handler, - Some(StageHandler::Prompt | StageHandler::Agent) - ) - { - if let Some(text) = outcome.output.get("text").and_then(Value::as_str) { - stage.response = Some(text.to_string()); - } - } - // An agent's answer: the `response.` the step wrote - // into the run context, as the prompt step writes it. - if stage.handler == Some(StageHandler::Agent) { - let response = node_name - .as_deref() - .and_then(|name| { - outcome - .context_updates - .get(format!("response.{name}").as_str()) - }) - .and_then(Value::as_str) - .or_else(|| outcome.output.as_str()); - if let Some(response) = response { - stage.response = Some(response.to_string()); - } - } - stage.live_streaming = Some(false); - apply_metrics(stage, &outcome.metrics); - if is_final { - stage.completion = Some(StageCompletion { - outcome: stage_outcome(&outcome.status), - notes: None, - failure_reason: failure_message(&outcome.status), - timestamp: at, - }); - stage.termination = Some(match outcome.status { - Status::TimedOut => fabro_types::CommandTermination::TimedOut, - Status::Cancelled => fabro_types::CommandTermination::Cancelled, - Status::Success - | Status::PartialSuccess { .. } - | Status::Failure(_) - | Status::Skipped => fabro_types::CommandTermination::Exited, - }); - } else { - stage.state = StageState::Retrying; - debug!(attempt = attempt.raw(), "attempt returned; a retry follows"); - } - } - } - Event::ControlRequested { .. } => { - if let Some(Derived::ControlRequested { - deliverable: true, - answer: Some(answer), - }) = &event.derived - { - let firing_key = event.subject.as_ref().and_then(|subject| { - subject - .firing - .map(|firing| stage_key(execution.raw(), firing.raw())) - }); - self.close_questions(answer.question.as_deref(), firing_key.as_deref(), at); - } - } - // ── Sandbox: the instance (VIEWS.md "Sandbox") ────────────────── - // The run's sandbox is the root invocation's scope. A child - // invocation's scope (a parallel branch) shares or owns another - // one and is not the run's; a re-acquisition (a resume, a - // replaced sandbox) names the current instance. - Event::ScopeAcquired { - sandbox, - duration_ms, - .. - } => { - if let Some(projection) = self.root_scope_projection(event) { - let plan = sandbox_plan_of(projection); - projection.sandbox = Some(RunSandbox::ready( - plan.clone(), - sandbox_instance(&plan, sandbox, *duration_ms), - )); - } - } - Event::ScopeFailed { - provider, - error, - causes, - duration_ms, - .. - } => { - if let Some(projection) = self.root_scope_projection(event) { - let plan = sandbox_plan_of(projection); - let provider = provider - .as_deref() - .and_then(provider_kind) - .unwrap_or_else(|| plan.provider.clone()); - projection.sandbox = Some(RunSandbox::failed(plan, RunSandboxFailure { - provider: provider.to_string(), - error: error.clone(), - causes: causes.clone(), - duration_ms: *duration_ms, - })); - } - } - Event::ExecutionStarted { .. } - | Event::TokenEmitted { .. } - | Event::RoutingResolved { .. } - | Event::RouteApplied { .. } - | Event::RetryElapsed { .. } - | Event::NodeExpanded { .. } - | Event::CancelRequested { .. } - | Event::KillRequested { .. } => {} - } - } - - /// The projection, when `event` is a scope record of the root - /// invocation: the run's own sandbox, not a child invocation's. - fn root_scope_projection(&mut self, event: &RunEvent) -> Option<&mut RunProjection> { - let root = self.state.root?; - if event.context.invocation.map(InvocationId::raw) != Some(root) { - return None; - } - self.projection.as_mut() - } - - fn fold_progress( - &mut self, - execution: ExecutionId, - event: &RunEvent, - ev: &StepEvent, - at: DateTime, - ) { - match ev { - StepEvent::Log { line, .. } => { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - let output = stage.output.get_or_insert_default(); - output.push_str(line); - output.push('\n'); - stage.output_bytes = Some(output.len() as u64); - stage.live_streaming = Some(true); - } - } - StepEvent::Artifact { .. } => {} - StepEvent::Custom(payload) => { - if let Some(parsed) = event.parsed() { - self.fold_parsed(execution, event, parsed, at); - return; - } - let kind = payload.get("kind").and_then(Value::as_str).unwrap_or(""); - match kind { - "pebble" => self.fold_pebble(execution, event, payload, at), - "attractor.prompt" => { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - stage.prompt = payload - .get("prompt") - .and_then(Value::as_str) - .map(str::to_string); - let model = payload.get("model").and_then(Value::as_str); - if let Some(model) = model { - let (provider, model_id) = split_model(model); - stage.provider_used = Some(StageModelUsage { - mode: StageModelUsage::MODE_PROMPT.to_string(), - provider: provider.map(str::to_string), - model: Some(model_id.to_string()), - reasoning_effort: None, - speed: None, - }); - stage.model = model_ref(provider, model_id); - } - } - } - "attractor.prompt.completed" => { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - stage.response = payload - .get("response") - .and_then(Value::as_str) - .map(str::to_string); - if let Some(usage) = usage_of(payload.get("usage")) { - stage.usage = usage; - } - } - } - "attractor.fallback.plan" => { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - let route = payload - .get("routes") - .and_then(Value::as_array) - .and_then(|routes| routes.first()); - if let Some(route) = route { - let provider = route.get("provider").and_then(Value::as_str); - let model = route.get("model").and_then(Value::as_str); - stage.provider_used = Some(StageModelUsage { - mode: StageModelUsage::MODE_AGENT.to_string(), - provider: provider.map(str::to_string), - model: model.map(str::to_string), - reasoning_effort: None, - speed: None, - }); - if let Some(model) = model { - stage.model = model_ref(provider, model); - } - } - } - } - // The tools a native session was offered, once per - // session (VIEWS.md "Agent activity", tools available): - // the stage's list is the union over its sessions, by - // name, in the order the sessions listed them. - "attractor.tools" => { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - let tools = payload.get("tools").and_then(Value::as_array); - for tool in tools.into_iter().flatten() { - let Some(summary) = tool_summary(tool) else { - continue; - }; - if !stage - .agent_tools - .iter() - .any(|known| known.name == summary.name) - { - stage.agent_tools.push(summary); - } - } - } - } - "attractor.parallel.branch.started" => { - let invocation = payload.get("invocation").and_then(Value::as_u64); - let index = payload - .get("index") - .and_then(Value::as_u64) - .and_then(|index| u32::try_from(index).ok()); - let fork_firing = payload - .get("occurrence") - .and_then(|occurrence| occurrence.get("firing")) - .and_then(Value::as_u64); - if let (Some(invocation), Some(index), Some(fork_firing)) = - (invocation, index, fork_firing) - { - let group = self - .state - .stages - .get(&stage_key(execution.raw(), fork_firing)) - .map(|stage| stage.stage_id.clone()); - if let Some(group) = group { - self.state.invocations.entry(invocation).or_default().branch = - Some((group, index)); - } - } - } - _ => {} - } - } - } - } - - fn fold_parsed( - &mut self, - execution: ExecutionId, - event: &RunEvent, - parsed: &Parsed, - at: DateTime, - ) { - match parsed { - Parsed::Question { question } => { - let Some(subject) = event.subject.as_ref() else { - return; - }; - let Some(firing) = subject.firing else { - return; - }; - let key = stage_key(execution.raw(), firing.raw()); - let label = self.state.stages.get(&key).map_or_else( - || subject.node.name.to_string(), - |stage| stage.stage_id.to_string(), - ); - self.state.questions.insert(question.id.clone(), key); - let Some(projection) = self.projection.as_mut() else { - return; - }; - projection - .pending_interviews - .insert(question.id.clone(), PendingInterviewRecord { - question: InterviewQuestionRecord { - id: question.id.clone(), - text: question.text.clone(), - stage: label, - question_type: question_type(question), - options: question - .options - .iter() - .map(|option| InterviewOption { - key: option.key.clone(), - label: option.label.clone(), - description: option.description.clone(), - preview: option.preview.clone(), - }) - .collect(), - allow_freeform: question.freeform, - timeout_seconds: question - .timeout_ms - .map(|timeout| timeout as f64 / 1000.0), - context_display: question.context.clone(), - review_target: question.reference.as_ref().and_then(review_target), - }, - started_at: at, - }); - apply_status( - projection, - RunStatus::Blocked { - blocked_reason: BlockedReason::HumanInputRequired, - }, - at, - ); - } - Parsed::QuestionExpired { expired } => { - self.close_questions(Some(expired.question.as_str()), None, at); - } - Parsed::Note { .. } => {} - } - } - - /// Close one question by id, or every question of a firing, and unblock - /// the run when none is left. - fn close_questions( - &mut self, - question: Option<&str>, - firing_key: Option<&str>, - at: DateTime, - ) { - let closed: Vec = match (question, firing_key) { - (Some(question), _) => vec![question.to_string()], - (None, Some(key)) => self - .state - .questions - .iter() - .filter(|(_, asked_by)| asked_by.as_str() == key) - .map(|(id, _)| id.clone()) - .collect(), - (None, None) => Vec::new(), - }; - for id in &closed { - self.state.questions.remove(id); - } - let Some(projection) = self.projection.as_mut() else { - return; - }; - for id in &closed { - projection.pending_interviews.remove(id); - } - if projection.pending_interviews.is_empty() - && matches!(projection.status, RunStatus::Blocked { .. }) - { - apply_status(projection, RunStatus::Running, at); - } - } - - fn fold_pebble( - &mut self, - execution: ExecutionId, - event: &RunEvent, - payload: &Value, - at: DateTime, - ) { - let Some(envelope) = payload.get("event") else { - return; - }; - let envelope: CodingAgentEvent = match serde_json::from_value(envelope.clone()) { - Ok(envelope) => envelope, - Err(error) => { - debug!(error = %error, "a pebble envelope did not decode; skipped"); - return; - } - }; - let Some(stage) = self.stage_of(execution, event.subject.as_ref()) else { - return; - }; - let agent = stage.agent.get_or_insert_default(); - agent.apply(&envelope); - if stage.completion.is_none() { - stage.usage = agent.usage.saturating_add(agent.descendant_usage()); - } - // A tool the stage's list names was called, by any of its sessions. - if let CodingEvent::ToolCallStarted { tool_name, .. } = &envelope.event { - if let Some(tool) = stage - .agent_tools - .iter_mut() - .find(|tool| tool.name == *tool_name) - { - tool.invoked = true; - } - } - let is_root = envelope.parent_session_id.is_none(); - #[expect( - clippy::wildcard_enum_match_arm, - reason = "pebble's event vocabulary is non-exhaustive and only some events project" - )] - match &envelope.event { - CodingEvent::SessionStarted { - provider, model, .. - } if is_root => { - stage.provider_used = Some(StageModelUsage { - mode: StageModelUsage::MODE_AGENT.to_string(), - provider: provider.clone(), - model: model.clone(), - reasoning_effort: None, - speed: None, - }); - if let Some(model) = model.as_deref() { - stage.model = model_ref(provider.as_deref(), model); - } - } - CodingEvent::LlmRequestStarted { requested_model } if is_root => { - stage.inference = Some(StageInferenceProjection { - session_id: envelope.session_id.clone(), - started_at: at, - requested_model: requested_model.clone(), - first_output_at: None, - first_output_kind: None, - retries: 0, - }); - } - CodingEvent::LlmFirstOutput { kind } => { - if let Some(inference) = stage.inference.as_mut() { - if inference.session_id == envelope.session_id { - inference.first_output_at = Some(at); - inference.first_output_kind = Some(*kind); - } - } - } - CodingEvent::LlmRetry { .. } => { - if let Some(inference) = stage.inference.as_mut() { - if inference.session_id == envelope.session_id { - inference.retries = inference.retries.saturating_add(1); - inference.first_output_at = None; - inference.first_output_kind = None; - } - } - } - CodingEvent::AssistantMessage { model, .. } => { - if is_root { - if let Some(provider) = stage - .provider_used - .as_ref() - .and_then(|used| used.provider.as_deref()) - { - stage.model = model_ref(Some(provider), model); - } - } - close_inference(stage, &envelope.session_id, at); - } - CodingEvent::Error { .. } | CodingEvent::RoundInterrupted { .. } => { - close_inference(stage, &envelope.session_id, at); - } - CodingEvent::SessionEnded => { - close_inference(stage, &envelope.session_id, at); - stage.close_tool_batch_for_session(&envelope.session_id, at); - } - CodingEvent::ToolCallStarted { tool_call_id, .. } if is_root => { - stage.open_tool_call(envelope.session_id.clone(), tool_call_id.clone(), at); - } - CodingEvent::ToolCallCompleted { tool_call_id, .. } if is_root => { - stage.close_tool_call(&envelope.session_id, tool_call_id, at); - } - _ => {} - } - } - - fn fold_view(&mut self, view: &ViewEvent, event: &RunEvent, at: DateTime) { - let Some(execution) = event.context.execution else { - return; - }; - match view { - ViewEvent::VisitStarted { .. } => { - let Some(subject) = event.subject.as_ref() else { - return; - }; - self.start_visit(execution, subject, at); - } - ViewEvent::WaitStateChanged { state } => { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - match state { - WaitState::AwaitingAdmission => { - if stage.state == StageState::Running { - stage.state = StageState::Pending; - } - } - WaitState::Running | WaitState::AwaitingAnswer | WaitState::Cancelling => { - stage.state = StageState::Running; - } - WaitState::AwaitingRetry => stage.state = StageState::Retrying, - } - } - } - ViewEvent::RetryScheduled { .. } => { - if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { - stage.state = StageState::Retrying; - } - } - ViewEvent::VisitCompleted { - outcome, - executed, - attempts, - } => { - let Some(stage) = self.stage_of(execution, event.subject.as_ref()) else { - return; - }; - stage.state = match outcome.status { - Status::Success => StageState::Succeeded, - Status::PartialSuccess { .. } => StageState::PartiallySucceeded, - Status::Failure(_) | Status::TimedOut => StageState::Failed, - Status::Skipped => StageState::Skipped, - Status::Cancelled => StageState::Cancelled, - }; - if stage.completion.is_none() || !*executed { - stage.completion = Some(StageCompletion { - outcome: stage_outcome(&outcome.status), - notes: None, - failure_reason: failure_message(&outcome.status), - timestamp: at, - }); - } - if stage.timing.is_none() { - let wall = stage - .started_at - .map_or(0, |started| timing::elapsed_ms(started, at)); - stage.set_authoritative_timing(StageTiming::new(wall, 0, 0)); - } - debug!(attempts, "visit completed"); - } - ViewEvent::ForkCompleted { - occurrence, - results, - .. - } => { - let key = stage_key(occurrence.execution.raw(), occurrence.firing.raw()); - let Some(stage_id) = self - .state - .stages - .get(&key) - .map(|stage| stage.stage_id.clone()) - else { - return; - }; - let Some(projection) = self.projection.as_mut() else { - return; - }; - if let Some(stage) = projection.stage_mut(&stage_id) { - stage.parallel_results = Some( - results - .iter() - .map(|result| ParallelBranchResult { - id: result.node.name.to_string(), - index: Some(result.branch.index as usize), - item_label: None, - status: stage_outcome(&result.status), - context_updates: BTreeMap::new(), - }) - .collect(), - ); - } - } - ViewEvent::ForkStarted { .. } - | ViewEvent::BranchCompleted { .. } - | ViewEvent::RunStalled { .. } => {} - } - } - - /// A firing exists: register its stage and, when it is a logical stage, - /// show it. - fn start_visit(&mut self, execution: ExecutionId, subject: &Subject, at: DateTime) { - let Some(firing) = subject.firing else { - return; - }; - let key = stage_key(execution.raw(), firing.raw()); - if self.state.stages.contains_key(&key) { - return; - } - let node_name = subject.node.name.to_string(); - let visit = visit_of(subject); - let meta_kind = node_meta_kind(&subject.node); - let shown = is_shown(&subject.node); - // Only a shown stage takes a label: a lowering node (a branch's - // parent-side delegate shares its target's name) never competes with - // the stage it stands for. - let mut stage_id = StageId::new(node_name.clone(), visit); - if shown { - stage_id = stage_label(&node_name, visit, execution, &self.state.labels); - self.state.labels.insert(stage_id.to_string()); - } - self.state.stages.insert(key, StageRef { - stage_id: stage_id.clone(), - shown, - node_name, - visit, - }); - if !shown { - return; - } - let branch = self - .state - .executions - .get(&execution.raw()) - .and_then(|invocation| self.state.invocations.get(invocation)) - .and_then(|invocation| invocation.branch.clone()); - let Some(projection) = self.projection.as_mut() else { - return; - }; - let since_created = at - .signed_duration_since(projection.spec.run_id.created_at()) - .num_milliseconds() - .max(0); - let ordinal = u32::try_from(since_created) - .unwrap_or(u32::MAX - 1) - .saturating_add(1); - let stage = projection.stage_entry(stage_id.node_id(), visit, first_event_seq(ordinal)); - stage.handler = Some(StageHandler::from_handler_type(Some(meta_kind))); - stage.started_at = Some(at); - stage.graph_visit = Some(visit); - stage.state = StageState::Pending; - stage.parallel_branch_id = branch.map(|(group, index)| ParallelBranchId::new(group, index)); - } - - /// The shown stage an event's subject firing belongs to. - fn stage_of( - &mut self, - execution: ExecutionId, - subject: Option<&Subject>, - ) -> Option<&mut StageProjection> { - let firing = subject?.firing?; - let stage = self - .state - .stages - .get(&stage_key(execution.raw(), firing.raw()))?; - if !stage.shown { - return None; - } - let stage_id = stage.stage_id.clone(); - self.projection.as_mut()?.stage_mut(&stage_id) - } -} - -impl Default for RunView { - fn default() -> Self { - Self::new() - } -} - -// ── Lifecycle ─────────────────────────────────────────────────────────── - -fn fold_lifecycle(projection: &mut RunProjection, record: &RunLifecycleRecord, at: DateTime) { - use RunLifecycleKind as Kind; - match record.transition { - Kind::Submitted => apply_status(projection, RunStatus::Submitted, at), - Kind::StartRequested => {} - Kind::Unpaused => { - let status = match projection.status { - RunStatus::Paused { - prior_block: Some(blocked_reason), - } => RunStatus::Blocked { blocked_reason }, - _ => RunStatus::Running, - }; - apply_status(projection, status, at); - if projection.pending_control == Some(RunControlAction::Unpause) { - projection.pending_control = None; - } - } - Kind::Pending => { - if let Some(status) = record.status { - apply_status(projection, status, at); - } - projection.approval = Some(RunApproval { - state: RunApprovalState::Pending, - requested_at: at, - decided_at: None, - denial_reason: None, - }); - } - Kind::Approved => { - if let Some(approval) = projection.approval.as_mut() { - approval.state = RunApprovalState::Approved; - approval.decided_at = Some(at); - } - } - Kind::Denied => { - if let Some(approval) = projection.approval.as_mut() { - approval.state = RunApprovalState::Denied; - approval.decided_at = Some(at); - approval.denial_reason.clone_from(&record.reason); - } - apply_status( - projection, - RunStatus::Failed { - reason: FailureReason::ApprovalDenied, - }, - at, - ); - } - Kind::Runnable => { - // A run left in flight by a restart goes back to the queue: the - // resume's `runnable` steps back from wherever the run stood. - let in_flight = matches!( - projection.status, - RunStatus::Starting - | RunStatus::Running - | RunStatus::Blocked { .. } - | RunStatus::Paused { .. } - ); - if in_flight && record.status == Some(RunStatus::Runnable) { - projection.status = RunStatus::Runnable; - projection.status_updated_at = at; - } else if let Some(status) = record.status { - apply_status(projection, status, at); - } - } - Kind::Blocked => { - // A block that lands while the run is paused waits behind the - // pause: the unpause restores it. - match (projection.status, record.status) { - (RunStatus::Paused { .. }, Some(RunStatus::Blocked { blocked_reason })) => { - apply_status( - projection, - RunStatus::Paused { - prior_block: Some(blocked_reason), - }, - at, - ); - } - (_, Some(status)) => apply_status(projection, status, at), - (_, None) => {} - } - } - Kind::Unblocked => { - let status = match projection.status { - RunStatus::Paused { .. } => RunStatus::Paused { prior_block: None }, - _ => record.status.unwrap_or(RunStatus::Running), - }; - apply_status(projection, status, at); - } - Kind::Starting | Kind::Running | Kind::Removing | Kind::Dead => { - if let Some(status) = record.status { - apply_status(projection, status, at); - } - } - Kind::Paused => { - let prior_block = match projection.status { - RunStatus::Blocked { blocked_reason } => Some(blocked_reason), - RunStatus::Paused { prior_block } => prior_block, - _ => None, - }; - apply_status(projection, RunStatus::Paused { prior_block }, at); - if projection.pending_control == Some(RunControlAction::Pause) { - projection.pending_control = None; - } - } - Kind::Succeeded | Kind::Failed => { - if let Some(status) = record.status { - apply_status(projection, status, at); - } - projection.pending_control = None; - if projection.conclusion.is_none() { - let (outcome, failure) = match record.status { - Some(RunStatus::Failed { reason }) => ( - StageOutcome::Failed { - retry_requested: false, - }, - Some(RunFailure { - reason, - detail: FailureDetail::new( - record - .reason - .clone() - .unwrap_or_else(|| "the run failed".to_string()), - FailureCategory::Deterministic, - ), - }), - ), - _ => (StageOutcome::Succeeded, None), - }; - projection.conclusion = Some(Conclusion { - timestamp: at, - status: outcome, - timing: RunTiming::default(), - failure, - final_git_commit_sha: None, - stages: Vec::new(), - usage: None, - total_retries: 0, - diff: RunDiff::default(), - }); - } - } - Kind::CancelRequested => projection.pending_control = Some(RunControlAction::Cancel), - Kind::PauseRequested => projection.pending_control = Some(RunControlAction::Pause), - Kind::UnpauseRequested => projection.pending_control = Some(RunControlAction::Unpause), - } -} - -/// Apply a status transition; one the lifecycle refuses is logged and -/// skipped, since the view never fails the run. -fn apply_status(projection: &mut RunProjection, status: RunStatus, at: DateTime) { - if let Err(error) = projection.try_apply_status(status, at) { - debug!(error = %error, "status transition not applied to the Petri projection"); - } -} - -fn touch(projection: &mut RunProjection, at: DateTime) { - if at > projection.last_event_at { - projection.last_event_at = at; - } -} - -// ── Helpers ───────────────────────────────────────────────────────────── - -/// The key of a stage: its execution and firing. -#[must_use] -pub fn stage_key(execution: u64, firing: u64) -> String { - format!("{execution}:{firing}") -} - -/// Which firing of its node a subject is, 1-based. -#[must_use] -pub fn visit_of(subject: &Subject) -> u32 { - subject.visit.unwrap_or(1).max(1) -} - -/// The role a frontend gave a node under `meta.kind`, or the empty string. -fn node_meta_kind(node: &NodeRef) -> &str { - node.meta.get("kind").and_then(Value::as_str).unwrap_or("") -} - -/// Whether a node is a logical stage the projection shows, or a lowering -/// node it keeps off the list: one a frontend marked synthetic, or a -/// parallel branch's delegate. -#[must_use] -pub fn is_shown(node: &NodeRef) -> bool { - let synthetic = node - .meta - .get("synthetic") - .and_then(Value::as_bool) - .unwrap_or(false); - !synthetic && node_meta_kind(node) != "parallel.branch" -} - -/// The label a shown firing takes, which is the stage id the projection -/// keys it by: `node@visit`, or `node/e@visit` when another -/// execution's firing already took that label. `taken` is every label given -/// so far; the caller adds the one returned. The interview adapter labels a -/// question's stage through this same rule, so the stage a question names -/// is the stage the projection shows. -#[must_use] -pub fn stage_label( - node_name: &str, - visit: u32, - execution: ExecutionId, - taken: &BTreeSet, -) -> StageId { - let stage_id = StageId::new(node_name.to_string(), visit); - if taken.contains(&stage_id.to_string()) { - return StageId::new(format!("{node_name}/e{}", execution.raw()), visit); - } - stage_id -} - -/// The fork firing and branch index a branch child's call slot names: -/// `branch:@::`. -fn branch_slot(slot: &str) -> Option<(u64, u32)> { - let rest = slot.strip_prefix("branch:")?; - let mut parts = rest.splitn(3, ':'); - let fork = parts.next()?; - let index = parts.next()?.parse::().ok()?; - let firing = fork.rsplit_once('@')?.1.parse::().ok()?; - Some((firing, index)) -} - -fn millis(recorded_at: u64) -> DateTime { - Utc.timestamp_millis_opt(i64::try_from(recorded_at).unwrap_or(i64::MAX)) - .single() - .unwrap_or_default() -} - -fn sandbox_plan(settings: &RunEnvironmentSettings) -> RunSandboxPlan { - RunSandboxPlan { - provider: settings.provider.clone(), - image: (settings.provider == SandboxProviderKind::DOCKER) - .then(|| settings.image.docker.clone()) - .flatten() - .filter(|image| !image.is_empty()), - snapshot: None, - } -} - -/// The plan the projection's sandbox carries, or the one its environment -/// settings give when no sandbox was projected yet. -fn sandbox_plan_of(projection: &RunProjection) -> RunSandboxPlan { - projection.sandbox.as_ref().map_or_else( - || sandbox_plan(&projection.spec.settings.run.environment), - |sandbox| sandbox.plan().clone(), - ) -} - -/// Fabro's name for the provider Petri's `scope.acquired` names: Petri's -/// `host` is Fabro's `local`; every other kind is spelled the same. `None` -/// for a name that is no provider kind. -fn provider_kind(provider: &str) -> Option { - if provider == "host" { - return Some(SandboxProviderKind::LOCAL); - } - SandboxProviderKind::try_new(provider).ok() -} - -/// The run's sandbox instance from Petri's record of the scope's -/// acquisition: the provider, the provider's id for the sandbox (what a -/// reconnect attaches by), its image and snapshot when the provider knows -/// them, the working directory, and how long the acquisition took. The -/// clone fields stay unset: Petri's checkout copies the bound repository -/// into the workspace and is not a clone Fabro made, and the workspace -/// roots are the provider's own layout, read live. `retained` waits for -/// the scope's release. -fn sandbox_instance( - plan: &RunSandboxPlan, - sandbox: &SandboxInstance, - ready_duration_ms: u64, -) -> RunSandboxInstance { - RunSandboxInstance { - provider: provider_kind(&sandbox.provider) - .unwrap_or_else(|| plan.provider.clone()), - image: sandbox - .image - .as_ref() - .map(ToString::to_string) - .or_else(|| plan.image.clone()), - snapshot: sandbox.snapshot.as_ref().map(ToString::to_string), - runtime: RunSandboxRuntime { - id: sandbox.instance.to_string(), - working_directory: sandbox.working_directory.to_string(), - repo_cloned: None, - clone_origin_url: None, - clone_branch: None, - workspace_root: None, - repos_root: None, - primary_repo_path: None, - primary_repo_link: None, - }, - ready_duration_ms: Some(ready_duration_ms), - retained: None, - } -} - -fn stage_outcome(status: &Status) -> StageOutcome { - match status { - Status::Success => StageOutcome::Succeeded, - Status::PartialSuccess { .. } => StageOutcome::PartiallySucceeded, - Status::Failure(info) => StageOutcome::Failed { - retry_requested: info.class.as_str() == "retry_requested", - }, - Status::Skipped => StageOutcome::Skipped, - Status::Cancelled | Status::TimedOut => StageOutcome::Failed { - retry_requested: false, - }, - } -} - -fn failure_message(status: &Status) -> Option { - match status { - Status::Failure(info) - | Status::PartialSuccess { - underlying: Some(info), - } => Some(info.message.clone()), - Status::TimedOut => Some("the step timed out".to_string()), - Status::Cancelled => Some("the step was cancelled".to_string()), - Status::Success | Status::PartialSuccess { underlying: None } | Status::Skipped => None, - } -} - -/// The finished attempt's metrics onto its stage: the timing and the usage -/// the backend reported. -/// One tool of an `attractor.tools` payload as the stage's list carries -/// it: the name and description as recorded, Pebble's `source` as it is, -/// and Pebble's behavioural category where Petri's says which (a -/// sub-agent tool); every other tool is `other`, because the payload -/// carries Petri's origin category (`builtin`, `mcp`, `host`, `question`), -/// not Pebble's permission class. `invoked` starts false and flips on the -/// session's `ToolCallStarted`. -fn tool_summary(tool: &Value) -> Option { - let name = tool.get("name").and_then(Value::as_str)?; - let description = tool - .get("description") - .and_then(Value::as_str) - .unwrap_or_default(); - let source = tool - .get("source") - .cloned() - .and_then(|source| serde_json::from_value::(source).ok()) - .unwrap_or_default(); - let category = match tool.get("category").and_then(Value::as_str) { - Some("subagent") => ToolCategory::Subagent, - _ => ToolCategory::Other, - }; - Some(ToolSummary { - name: name.to_string(), - description: description.to_string(), - source, - category, - invoked: false, - }) -} - -/// The question's `reference` as Fabro's review target, when it is one -/// Fabro's validation admits (a `document`, or a reference without a kind, -/// with a label and an absolute HTTP URL within Fabro's limits). -fn review_target(reference: &QuestionReference) -> Option { - let kind = match reference.kind.as_deref() { - Some("document") | None => ReviewTargetKind::Document, - Some(_) => return None, - }; - ReviewTarget::new(reference.label.clone(), reference.url.clone(), kind).ok() -} - -fn apply_metrics(stage: &mut StageProjection, metrics: &Metrics) { - let custom = &metrics.custom; - let inference = custom - .get("pebble.inference_ms") - .and_then(Value::as_u64) - .unwrap_or(0); - let tool = custom - .get("pebble.tool_ms") - .and_then(Value::as_u64) - .unwrap_or(0); - let wall = metrics.duration_ms.unwrap_or(0); - let (inference, tool) = match stage.handler { - Some(StageHandler::Prompt) => (wall, 0), - Some(StageHandler::Command) => (0, wall), - _ => (inference, tool), - }; - stage.set_authoritative_timing(StageTiming::new(wall, inference, tool).clamped_to_wall()); - if let Some(usage) = - usage_of(custom.get("pebble.usage")).or_else(|| usage_of(custom.get("prompt.usage"))) - { - stage.usage = usage; - } - if let Some(sessions) = custom - .get("pebble.subagents") - .and_then(|subagents| subagents.get("sessions")) - .and_then(Value::as_array) - { - let mut by_model: Vec = Vec::new(); - for session in sessions { - let provider = session.get("provider").and_then(Value::as_str); - let model = session.get("model").and_then(Value::as_str); - let Some(usage) = usage_of(session.get("usage")) else { - continue; - }; - let Some(model) = model.and_then(|model| model_ref(provider, model)) else { - continue; - }; - if let Some(entry) = by_model.iter_mut().find(|entry| entry.model == model) { - entry.usage = entry.usage.saturating_add(usage); - } else { - by_model.push(ModelUsage::new(model, usage)); - } - } - if !by_model.is_empty() { - stage.usage_by_model = by_model; - } - } -} - -fn usage_of(value: Option<&Value>) -> Option { - serde_json::from_value(value?.clone()).ok() -} - -/// `provider/model` into its parts, or the model alone. -fn split_model(model: &str) -> (Option<&str>, &str) { - match model.split_once('/') { - Some((provider, model)) if !provider.is_empty() && !model.is_empty() => { - (Some(provider), model) - } - _ => (None, model), - } -} - -fn model_ref(provider: Option<&str>, model: &str) -> Option { - let provider = provider.filter(|provider| !provider.is_empty())?; - Some(ModelRef::new( - ProviderId::new(provider), - ModelId::new(model), - )) -} - -fn close_inference(stage: &mut StageProjection, session_id: &str, at: DateTime) { - let open = stage - .inference - .as_ref() - .is_some_and(|inference| inference.session_id == session_id); - if !open { - return; - } - if let Some(inference) = stage.inference.take() { - stage.accumulate_inference_ms(timing::elapsed_ms(inference.started_at, at)); - } -} - -/// The run id a Petri run key names. -#[must_use] -pub fn run_id_of(key: &str) -> Option { - key.parse().ok() -} - -#[cfg(test)] -mod tests { - use fabro_store::platform_records::RunCreatedRecord; - use fabro_types::test_support as types_support; - use petri_runtime::driver::BranchRole; - use petri_runtime::ir::{FiringId, NodeId}; - - use super::*; - - #[test] - fn a_branch_slot_names_the_fork_firing_and_the_index() { - assert_eq!(branch_slot("branch:fan@7:2:review"), Some((7, 2))); - assert_eq!(branch_slot("branch:fan@7:x:review"), None); - assert_eq!(branch_slot("child:0"), None); - } - - #[test] - fn a_model_selector_splits_into_provider_and_model() { - assert_eq!(split_model("openai/gpt-5.4"), (Some("openai"), "gpt-5.4")); - assert_eq!(split_model("gpt-5.4"), (None, "gpt-5.4")); - assert!(model_ref(None, "gpt-5.4").is_none()); - assert!(model_ref(Some("openai"), "gpt-5.4").is_some()); - } - - #[test] - fn a_taken_label_is_made_unique_by_the_execution() { - let mut view = RunView::new(); - let created = StoredPlatformRecord { - seq: 1, - recorded_at: 1_000, - record: PlatformRecord::RunCreated(RunCreatedRecord { - spec: types_support::test_run_spec(), - title: Some("A run".to_string()), - parent_id: None, - retried_from: None, - web_url: None, - }), - position: None, - }; - view.fold(&Item::Platform(&created), 1); - let subject = |name: &str| Subject { - node: NodeRef { - id: NodeId::new(1), - name: name.into(), - kind: "attractor/command".into(), - meta: serde_json::json!({ "kind": "command" }), - }, - firing: Some(FiringId::new(4)), - visit: Some(1), - attempt: None, - generation: None, - branch: BranchRole::None, - }; - view.start_visit(ExecutionId::new(1), &subject("build"), millis(2_000)); - view.start_visit(ExecutionId::new(2), &subject("build"), millis(3_000)); - let labels: Vec = view - .projection() - .expect("the run was created") - .iter_stages() - .map(|(id, _)| id.to_string()) - .collect(); - assert_eq!(labels, vec!["build@1", "build/e2@1"]); - assert_eq!(view.state.stages.len(), 2); - } -} diff --git a/lib/components/fabro-petri/src/projection/coordinator.rs b/lib/components/fabro-petri/src/projection/coordinator.rs new file mode 100644 index 000000000..ffa5093ce --- /dev/null +++ b/lib/components/fabro-petri/src/projection/coordinator.rs @@ -0,0 +1,256 @@ +//! Petri's coordinator events folded into the view: the run's start, its +//! invocations and executions, pause and unpause, its finish, and the +//! release of its sandbox (VIEWS.md "Run", "Sandbox"). + +use chrono::{DateTime, Utc}; +use fabro_types::{ + Conclusion, FailureCategory, FailureDetail, FailureReason, RunControlAction, RunFailure, + RunSandbox, RunStatus, RunTiming, StageOutcome, StartRecord, SuccessReason, usage_rollup, +}; +use petri_execution::CoordinatorEvent; +use petri_execution::events::RunEvent; + +use super::{InvocationRef, RunView, apply_status, stage_key}; + +impl RunView { + pub(super) fn fold_coordinator( + &mut self, + record: &CoordinatorEvent, + event: &RunEvent, + at: DateTime, + ) { + match record { + CoordinatorEvent::RunStarted { + root, forked_from, .. + } => { + self.state.root = Some(root.raw()); + self.state.started_at = Some(event.recorded_at); + if let Some(projection) = self.projection.as_mut() { + // A fork's declaration names its source; a parse failure + // means the source was not a Fabro run, which the + // projection cannot show. + projection.forked_from = forked_from.as_ref().and_then(|origin| { + Some(fabro_types::ForkOrigin { + source_run_id: origin.source.as_str().parse().ok()?, + execution: origin.position.execution.raw(), + firing: origin.position.firing.raw(), + rerun_last: origin.rerun_last, + }) + }); + apply_status(projection, RunStatus::Running, at); + projection.start = Some(StartRecord { + start_time: at, + run_branch: self.state.run_branch.clone(), + base_sha: self.state.base_sha.clone(), + }); + // The scope's sandbox is acquired next; `scope.acquired` + // or `scope.failed` settles it. + if let Some(sandbox) = projection.sandbox.take() { + projection.sandbox = Some(RunSandbox::initializing(sandbox.plan().clone())); + } + } + } + CoordinatorEvent::InvocationDeclared { invocation, .. } => { + let mut info = InvocationRef::default(); + if let Some(parent) = &event.context.parent { + info.parent = Some((parent.execution.raw(), parent.firing.raw())); + if let Some((fork_firing, index)) = branch_slot(&parent.slot) { + let group = self + .state + .stages + .get(&stage_key(parent.execution.raw(), fork_firing)) + .map(|stage| stage.stage_id.clone()); + if let Some(group) = group { + info.branch = Some((group, index)); + } + } + } + self.state.invocations.insert(invocation.raw(), info); + } + CoordinatorEvent::ExecutionDeclared { + execution, + invocation, + .. + } => { + self.state + .executions + .insert(execution.raw(), invocation.raw()); + } + CoordinatorEvent::InvocationFinished { invocation, result } => { + let info = self.state.invocations.entry(invocation.raw()).or_default(); + info.failure = result + .failure + .as_ref() + .map(|failure| failure.message.clone()); + info.output = Some(result.output.clone()); + } + CoordinatorEvent::RunPaused => { + if let Some(projection) = self.projection.as_mut() { + let prior_block = match projection.status { + RunStatus::Blocked { blocked_reason } => Some(blocked_reason), + _ => None, + }; + apply_status(projection, RunStatus::Paused { prior_block }, at); + if projection.pending_control == Some(RunControlAction::Pause) { + projection.pending_control = None; + } + } + } + CoordinatorEvent::RunUnpaused => { + if let Some(projection) = self.projection.as_mut() { + let next = match projection.status { + RunStatus::Paused { + prior_block: Some(blocked_reason), + } => RunStatus::Blocked { blocked_reason }, + _ => RunStatus::Running, + }; + apply_status(projection, next, at); + if projection.pending_control == Some(RunControlAction::Unpause) { + projection.pending_control = None; + } + } + } + CoordinatorEvent::RunFinished { status } => { + self.state.finished = Some(status.to_string()); + self.conclude(status.to_string().as_str(), at); + } + // ── Sandbox: the retention outcome (VIEWS.md "Sandbox") ───────── + // The instance stays on `Run.sandbox`: it names what ran, and + // `retained` says whether it still exists. + CoordinatorEvent::ScopeReleased { + invocation, + retained, + .. + } => { + if Some(invocation.raw()) == self.state.root { + self.state.sandbox_retained = Some(*retained); + if let Some(sandbox) = self + .projection + .as_mut() + .and_then(|projection| projection.sandbox.as_mut()) + { + sandbox.set_retained(*retained); + } + } + } + CoordinatorEvent::GraphRegistered { .. } + | CoordinatorEvent::ExecutionFinished { .. } + | CoordinatorEvent::InvocationCancelRequested { .. } + | CoordinatorEvent::RunNoteRecorded { .. } => {} + } + } + + /// The run's conclusion, from its recorded finish and what the stages + + /// The run's conclusion, from its recorded finish and what the stages + /// summed to. + fn conclude(&mut self, status: &str, at: DateTime) { + let Some(projection) = self.projection.as_mut() else { + return; + }; + let root = self + .state + .root + .and_then(|root| self.state.invocations.get(&root)); + let failure_message = root.and_then(|root| root.failure.clone()); + let (run_status, outcome, failure) = match status { + "success" => ( + RunStatus::Succeeded { + reason: SuccessReason::Completed, + }, + StageOutcome::Succeeded, + None, + ), + "cancelled" => ( + RunStatus::Failed { + reason: FailureReason::Cancelled, + }, + StageOutcome::Failed { + retry_requested: false, + }, + Some(RunFailure { + reason: FailureReason::Cancelled, + detail: FailureDetail::new( + failure_message + .clone() + .unwrap_or_else(|| "the run was cancelled".to_string()), + FailureCategory::Canceled, + ), + }), + ), + _ => ( + RunStatus::Failed { + reason: FailureReason::WorkflowError, + }, + StageOutcome::Failed { + retry_requested: false, + }, + Some(RunFailure { + reason: FailureReason::WorkflowError, + detail: FailureDetail::new( + failure_message + .clone() + .unwrap_or_else(|| "the run failed".to_string()), + FailureCategory::Deterministic, + ), + }), + ), + }; + apply_status(projection, run_status, at); + projection.pending_control = None; + projection.pending_interviews.clear(); + let rollup = usage_rollup::usage_rollup_from_projection(projection); + let (stages, total_retries) = rollup.conclusion_stages(projection); + let wall_time_ms = self.state.started_at.map_or(0, |started| { + u64::try_from(at.timestamp_millis()) + .unwrap_or(0) + .saturating_sub(started) + }); + let timing = RunTiming::new( + wall_time_ms, + rollup.timing.inference_time_ms, + rollup.timing.tool_time_ms, + ); + let last_checkpoint = projection.checkpoints.last(); + projection.conclusion = Some(Conclusion { + timestamp: at, + status: outcome, + timing, + failure, + final_git_commit_sha: last_checkpoint + .and_then(|checkpoint| checkpoint.checkpoint.git_commit_sha.clone()), + stages, + usage: rollup.usage_if_present(), + total_retries, + diff: self + .state + .run_diff + .clone() + .or_else(|| last_checkpoint.map(|checkpoint| checkpoint.diff.clone())) + .unwrap_or_default(), + }); + } +} + +/// The fork firing and branch index a branch child's call slot names: +/// `branch:@::`. +fn branch_slot(slot: &str) -> Option<(u64, u32)> { + let rest = slot.strip_prefix("branch:")?; + let mut parts = rest.splitn(3, ':'); + let fork = parts.next()?; + let index = parts.next()?.parse::().ok()?; + let firing = fork.rsplit_once('@')?.1.parse::().ok()?; + Some((firing, index)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_branch_slot_names_the_fork_firing_and_the_index() { + assert_eq!(branch_slot("branch:fan@7:2:review"), Some((7, 2))); + assert_eq!(branch_slot("branch:fan@7:x:review"), None); + assert_eq!(branch_slot("child:0"), None); + } +} diff --git a/lib/components/fabro-petri/src/projection/engine.rs b/lib/components/fabro-petri/src/projection/engine.rs new file mode 100644 index 000000000..a936ab494 --- /dev/null +++ b/lib/components/fabro-petri/src/projection/engine.rs @@ -0,0 +1,457 @@ +//! Petri's engine and view events folded into the stages: a firing's +//! visit, its attempts and their outcomes, its wait states, and the scope +//! its sandbox was acquired in (VIEWS.md "Stages", "Sandbox"). + +use std::collections::BTreeMap; + +use chrono::{DateTime, Utc}; +use fabro_types::{ + ModelUsage, ParallelBranchId, ParallelBranchResult, RunProjection, RunSandbox, + RunSandboxFailure, StageCompletion, StageHandler, StageId, StageOutcome, StageProjection, + StageState, StageTiming, first_event_seq, parse_blob_ref, timing, +}; +use petri_execution::events::{Derived, RunEvent, Subject, ViewEvent, WaitState}; +use petri_execution::{ExecutionId, InvocationId}; +use petri_runtime::engine::{Admission, Event}; +use petri_runtime::ir::{Metrics, Status}; +use serde_json::Value; +use tracing::debug; + +use super::model::{model_ref, usage_of}; +use super::sandbox::{provider_kind, sandbox_instance, sandbox_plan_of}; +use super::{RunView, StageRef, is_shown, node_meta_kind, stage_key, stage_label, visit_of}; + +impl RunView { + pub(super) fn fold_engine(&mut self, engine: &Event, event: &RunEvent, at: DateTime) { + let Some(execution) = event.context.execution else { + return; + }; + match engine { + Event::AdmissionDecided { decision, .. } => { + if let Admission::Skip { outcome } = decision { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + stage.state = StageState::Skipped; + stage.completion = Some(StageCompletion { + outcome: StageOutcome::Skipped, + notes: None, + failure_reason: failure_message(&outcome.status), + timestamp: at, + }); + } + } + } + Event::StepStarted { attempt, .. } => { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + if attempt.raw() > 1 { + stage.clear_live_timing(); + stage.output = None; + stage.output_bytes = None; + } + stage.state = StageState::Running; + stage.live_streaming = Some(true); + } + } + Event::StepProgressRecorded { ev, .. } => { + self.fold_progress(execution, event, ev, at); + } + Event::StepFinished { + firing, + attempt, + outcome, + } => { + self.state + .finished_firings + .insert(stage_key(execution.raw(), firing.raw())); + let is_final = matches!( + event.derived, + Some(Derived::StepFinished { is_final: true, .. }) + ); + let node_name = event + .subject + .as_ref() + .map(|subject| subject.node.name.to_string()); + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + // The step's output: a string, or a command's `stdout`, + // either of which is a `blob://` reference when the + // step offloaded it. The reference stays as it is; the + // bytes it names are the live log's. + let output = outcome + .output + .as_str() + .or_else(|| outcome.output.get("stdout").and_then(Value::as_str)); + if let Some(output) = output { + if parse_blob_ref(output).is_none() { + stage.output_bytes = Some(output.len() as u64); + } + stage.output = Some(output.to_string()); + } + // A simulated step (a dry run) answers with its text. + let simulated = outcome + .output + .get("simulated") + .and_then(Value::as_bool) + .unwrap_or(false); + if simulated + && matches!( + stage.handler, + Some(StageHandler::Prompt | StageHandler::Agent) + ) + { + if let Some(text) = outcome.output.get("text").and_then(Value::as_str) { + stage.response = Some(text.to_string()); + } + } + // An agent's answer: the `response.` the step wrote + // into the run context, as the prompt step writes it. + if stage.handler == Some(StageHandler::Agent) { + let response = node_name + .as_deref() + .and_then(|name| { + outcome + .context_updates + .get(format!("response.{name}").as_str()) + }) + .and_then(Value::as_str) + .or_else(|| outcome.output.as_str()); + if let Some(response) = response { + stage.response = Some(response.to_string()); + } + } + stage.live_streaming = Some(false); + apply_metrics(stage, &outcome.metrics); + if is_final { + stage.completion = Some(StageCompletion { + outcome: stage_outcome(&outcome.status), + notes: None, + failure_reason: failure_message(&outcome.status), + timestamp: at, + }); + stage.termination = Some(match outcome.status { + Status::TimedOut => fabro_types::CommandTermination::TimedOut, + Status::Cancelled => fabro_types::CommandTermination::Cancelled, + Status::Success + | Status::PartialSuccess { .. } + | Status::Failure(_) + | Status::Skipped => fabro_types::CommandTermination::Exited, + }); + } else { + stage.state = StageState::Retrying; + debug!(attempt = attempt.raw(), "attempt returned; a retry follows"); + } + } + } + Event::ControlRequested { .. } => { + if let Some(Derived::ControlRequested { + deliverable: true, + answer: Some(answer), + }) = &event.derived + { + let firing_key = event.subject.as_ref().and_then(|subject| { + subject + .firing + .map(|firing| stage_key(execution.raw(), firing.raw())) + }); + self.close_questions(answer.question.as_deref(), firing_key.as_deref(), at); + } + } + // ── Sandbox: the instance (VIEWS.md "Sandbox") ────────────────── + // The run's sandbox is the root invocation's scope. A child + // invocation's scope (a parallel branch) shares or owns another + // one and is not the run's; a re-acquisition (a resume, a + // replaced sandbox) names the current instance. + Event::ScopeAcquired { + sandbox, + duration_ms, + .. + } => { + if let Some(projection) = self.root_scope_projection(event) { + let plan = sandbox_plan_of(projection); + projection.sandbox = Some(RunSandbox::ready( + plan.clone(), + sandbox_instance(&plan, sandbox, *duration_ms), + )); + } + } + Event::ScopeFailed { + provider, + error, + causes, + duration_ms, + .. + } => { + if let Some(projection) = self.root_scope_projection(event) { + let plan = sandbox_plan_of(projection); + let provider = provider + .as_deref() + .and_then(provider_kind) + .unwrap_or_else(|| plan.provider.clone()); + projection.sandbox = Some(RunSandbox::failed(plan, RunSandboxFailure { + provider: provider.to_string(), + error: error.clone(), + causes: causes.clone(), + duration_ms: *duration_ms, + })); + } + } + Event::ExecutionStarted { .. } + | Event::TokenEmitted { .. } + | Event::RoutingResolved { .. } + | Event::RouteApplied { .. } + | Event::RetryElapsed { .. } + | Event::NodeExpanded { .. } + | Event::CancelRequested { .. } + | Event::KillRequested { .. } => {} + } + } + + /// The projection, when `event` is a scope record of the root + + /// The projection, when `event` is a scope record of the root + /// invocation: the run's own sandbox, not a child invocation's. + fn root_scope_projection(&mut self, event: &RunEvent) -> Option<&mut RunProjection> { + let root = self.state.root?; + if event.context.invocation.map(InvocationId::raw) != Some(root) { + return None; + } + self.projection.as_mut() + } + + pub(super) fn fold_view(&mut self, view: &ViewEvent, event: &RunEvent, at: DateTime) { + let Some(execution) = event.context.execution else { + return; + }; + match view { + ViewEvent::VisitStarted { .. } => { + let Some(subject) = event.subject.as_ref() else { + return; + }; + self.start_visit(execution, subject, at); + } + ViewEvent::WaitStateChanged { state } => { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + match state { + WaitState::AwaitingAdmission => { + if stage.state == StageState::Running { + stage.state = StageState::Pending; + } + } + WaitState::Running | WaitState::AwaitingAnswer | WaitState::Cancelling => { + stage.state = StageState::Running; + } + WaitState::AwaitingRetry => stage.state = StageState::Retrying, + } + } + } + ViewEvent::RetryScheduled { .. } => { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + stage.state = StageState::Retrying; + } + } + ViewEvent::VisitCompleted { + outcome, + executed, + attempts, + } => { + let Some(stage) = self.stage_of(execution, event.subject.as_ref()) else { + return; + }; + stage.state = match outcome.status { + Status::Success => StageState::Succeeded, + Status::PartialSuccess { .. } => StageState::PartiallySucceeded, + Status::Failure(_) | Status::TimedOut => StageState::Failed, + Status::Skipped => StageState::Skipped, + Status::Cancelled => StageState::Cancelled, + }; + if stage.completion.is_none() || !*executed { + stage.completion = Some(StageCompletion { + outcome: stage_outcome(&outcome.status), + notes: None, + failure_reason: failure_message(&outcome.status), + timestamp: at, + }); + } + if stage.timing.is_none() { + let wall = stage + .started_at + .map_or(0, |started| timing::elapsed_ms(started, at)); + stage.set_authoritative_timing(StageTiming::new(wall, 0, 0)); + } + debug!(attempts, "visit completed"); + } + ViewEvent::ForkCompleted { + occurrence, + results, + .. + } => { + let key = stage_key(occurrence.execution.raw(), occurrence.firing.raw()); + let Some(stage_id) = self + .state + .stages + .get(&key) + .map(|stage| stage.stage_id.clone()) + else { + return; + }; + let Some(projection) = self.projection.as_mut() else { + return; + }; + if let Some(stage) = projection.stage_mut(&stage_id) { + stage.parallel_results = Some( + results + .iter() + .map(|result| ParallelBranchResult { + id: result.node.name.to_string(), + index: Some(result.branch.index as usize), + item_label: None, + status: stage_outcome(&result.status), + context_updates: BTreeMap::new(), + }) + .collect(), + ); + } + } + ViewEvent::ForkStarted { .. } + | ViewEvent::BranchCompleted { .. } + | ViewEvent::RunStalled { .. } => {} + } + } + + /// A firing exists: register its stage and, when it is a logical stage, + + /// A firing exists: register its stage and, when it is a logical stage, + /// show it. + pub(super) fn start_visit( + &mut self, + execution: ExecutionId, + subject: &Subject, + at: DateTime, + ) { + let Some(firing) = subject.firing else { + return; + }; + let key = stage_key(execution.raw(), firing.raw()); + if self.state.stages.contains_key(&key) { + return; + } + let node_name = subject.node.name.to_string(); + let visit = visit_of(subject); + let meta_kind = node_meta_kind(&subject.node); + let shown = is_shown(&subject.node); + // Only a shown stage takes a label: a lowering node (a branch's + // parent-side delegate shares its target's name) never competes with + // the stage it stands for. + let mut stage_id = StageId::new(node_name.clone(), visit); + if shown { + stage_id = stage_label(&node_name, visit, execution, &self.state.labels); + self.state.labels.insert(stage_id.to_string()); + } + self.state.stages.insert(key, StageRef { + stage_id: stage_id.clone(), + shown, + node_name, + visit, + }); + if !shown { + return; + } + let branch = self + .state + .executions + .get(&execution.raw()) + .and_then(|invocation| self.state.invocations.get(invocation)) + .and_then(|invocation| invocation.branch.clone()); + let Some(projection) = self.projection.as_mut() else { + return; + }; + let since_created = at + .signed_duration_since(projection.spec.run_id.created_at()) + .num_milliseconds() + .max(0); + let ordinal = u32::try_from(since_created) + .unwrap_or(u32::MAX - 1) + .saturating_add(1); + let stage = projection.stage_entry(stage_id.node_id(), visit, first_event_seq(ordinal)); + stage.handler = Some(StageHandler::from_handler_type(Some(meta_kind))); + stage.started_at = Some(at); + stage.graph_visit = Some(visit); + stage.state = StageState::Pending; + stage.parallel_branch_id = branch.map(|(group, index)| ParallelBranchId::new(group, index)); + } +} + +pub(super) fn stage_outcome(status: &Status) -> StageOutcome { + match status { + Status::Success => StageOutcome::Succeeded, + Status::PartialSuccess { .. } => StageOutcome::PartiallySucceeded, + Status::Failure(info) => StageOutcome::Failed { + retry_requested: info.class.as_str() == "retry_requested", + }, + Status::Skipped => StageOutcome::Skipped, + Status::Cancelled | Status::TimedOut => StageOutcome::Failed { + retry_requested: false, + }, + } +} + +pub(super) fn failure_message(status: &Status) -> Option { + match status { + Status::Failure(info) + | Status::PartialSuccess { + underlying: Some(info), + } => Some(info.message.clone()), + Status::TimedOut => Some("the step timed out".to_string()), + Status::Cancelled => Some("the step was cancelled".to_string()), + Status::Success | Status::PartialSuccess { underlying: None } | Status::Skipped => None, + } +} + +/// The finished attempt's metrics onto its stage: the timing and the usage +/// the backend reported. +fn apply_metrics(stage: &mut StageProjection, metrics: &Metrics) { + let custom = &metrics.custom; + let inference = custom + .get("pebble.inference_ms") + .and_then(Value::as_u64) + .unwrap_or(0); + let tool = custom + .get("pebble.tool_ms") + .and_then(Value::as_u64) + .unwrap_or(0); + let wall = metrics.duration_ms.unwrap_or(0); + let (inference, tool) = match stage.handler { + Some(StageHandler::Prompt) => (wall, 0), + Some(StageHandler::Command) => (0, wall), + _ => (inference, tool), + }; + stage.set_authoritative_timing(StageTiming::new(wall, inference, tool).clamped_to_wall()); + if let Some(usage) = + usage_of(custom.get("pebble.usage")).or_else(|| usage_of(custom.get("prompt.usage"))) + { + stage.usage = usage; + } + if let Some(sessions) = custom + .get("pebble.subagents") + .and_then(|subagents| subagents.get("sessions")) + .and_then(Value::as_array) + { + let mut by_model: Vec = Vec::new(); + for session in sessions { + let provider = session.get("provider").and_then(Value::as_str); + let model = session.get("model").and_then(Value::as_str); + let Some(usage) = usage_of(session.get("usage")) else { + continue; + }; + let Some(model) = model.and_then(|model| model_ref(provider, model)) else { + continue; + }; + if let Some(entry) = by_model.iter_mut().find(|entry| entry.model == model) { + entry.usage = entry.usage.saturating_add(usage); + } else { + by_model.push(ModelUsage::new(model, usage)); + } + } + if !by_model.is_empty() { + stage.usage_by_model = by_model; + } + } +} diff --git a/lib/components/fabro-petri/src/projection/mod.rs b/lib/components/fabro-petri/src/projection/mod.rs new file mode 100644 index 000000000..4191bdd62 --- /dev/null +++ b/lib/components/fabro-petri/src/projection/mod.rs @@ -0,0 +1,344 @@ +//! The projection of a Petri run: Petri's public events and Fabro's platform +//! records folded into the view Fabro's read side serves. +//! +//! The fold is pure. [`RunView`] holds the [`RunProjection`] the API serves +//! (`GET /runs/{id}/state`, the run list through its summary) and the +//! bookkeeping the fold needs between items ([`FoldState`]): which Petri +//! firing each stage is, which invocation each execution belongs to and +//! whether it is a parallel branch, which stage asked each open question. +//! Both halves are stored by the projector and reloaded for the next pass, +//! so a pass folds only the items past the committed positions. +//! +//! The mapping follows `VIEWS.md`, row by row. The stage key is `(execution, +//! firing)`; Fabro's `StageId` (`node@visit`) is the display label the +//! `RunProjection` keys stages by, and a label two firings would share (two +//! child invocations with the same node name and visit) is made unique by +//! naming the execution. What the matrix leaves default is left default +//! here and named in the crate's README. +//! +//! Every item the fold sees carries the delivery sequence the projector +//! assigned it (`stream_seq`), which a checkpoint keeps as its `seq`. A +//! stage's `first_event_seq`, the key the stage list sorts by, is not the +//! delivery sequence: two logs' records can be committed in an order that +//! differs from their recording times by a few positions, and the view +//! built live must equal the view rebuilt from the records alone. It is the +//! milliseconds from the run's creation to the stage's `visit.started`, +//! plus one, which is the same however the records were delivered. + +mod coordinator; +mod engine; +mod model; +mod platform; +mod progress; +mod sandbox; + +use std::collections::{BTreeMap, BTreeSet}; + +use chrono::{DateTime, TimeZone as _, Utc}; +use fabro_store::platform_records::StoredPlatformRecord; +use fabro_types::{RunDiff, RunId, RunProjection, RunStatus, StageId, StageProjection}; +use petri_execution::ExecutionId; +use petri_execution::events::{NodeRef, RunEvent, Subject}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tracing::debug; + +/// One item the projector hands the fold, with its delivery sequence. +pub enum Item<'a> { + Petri(&'a RunEvent), + Platform(&'a StoredPlatformRecord), +} + +/// A stage as the fold knows it: its label in the projection, and what it +/// learned about it. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct StageRef { + pub stage_id: StageId, + /// Whether the stage is a logical one the projection shows, or a + /// lowering node it keeps off the list. + pub shown: bool, + /// The node's instance name and visit, for the collision rule. + pub node_name: String, + pub visit: u32, +} + +/// What the fold knows about one invocation. +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct InvocationRef { + /// The calling execution and firing, for a nested invocation. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent: Option<(u64, u64)>, + /// The parallel group and branch index, for a branch child. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub branch: Option<(StageId, u32)>, + /// The result the invocation recorded, for the root. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub failure: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub output: Option, +} + +/// Whether the run's durable record is whole, as the projector last read it. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RecordHealth { + pub complete: bool, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub incomplete: Vec, +} + +/// The fold's bookkeeping between items. +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct FoldState { + /// Stages by `":"`. + #[serde(default)] + pub stages: BTreeMap, + /// Labels taken, so a second firing with the same name and visit gets + /// its own. + #[serde(default)] + pub labels: BTreeSet, + #[serde(default)] + pub invocations: BTreeMap, + /// Which invocation each execution belongs to. + #[serde(default)] + pub executions: BTreeMap, + /// Open questions by id: the stage that asked. + #[serde(default)] + pub questions: BTreeMap, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub root: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at: Option, + /// The run's recorded finish, when Petri recorded one. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finished: Option, + /// The run branch and base sha, when they arrive before `run.started`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub run_branch: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub base_sha: Option, + #[serde(default)] + pub checkpoints: u32, + /// The run's diff as its `run.diff` record gave it, whichever side of + /// the run's finish it arrived on. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub run_diff: Option, + #[serde(default)] + pub health: RecordHealth, + /// Firings (`":"`) whose attempt has recorded a + /// finish: what a position-keyed platform record may be streamed + /// behind. + #[serde(default)] + pub finished_firings: BTreeSet, + /// Whether the run's sandbox still exists after its release + /// (`scope.released` `retained`): kept stopped, or deleted. Absent until + /// the root invocation's lease was released. The view carries the same + /// fact as `RunSandboxInstance.retained`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub sandbox_retained: Option, +} + +impl FoldState { + /// Whether Petri recorded the run's finish. + #[must_use] + pub fn finished_run(&self) -> bool { + self.finished.is_some() + } +} + +/// The view of one run: what the API serves and what the fold keeps. +#[derive(Clone, Debug)] +pub struct RunView { + pub projection: Option, + pub state: FoldState, +} + +impl RunView { + #[must_use] + pub fn new() -> Self { + Self { + projection: None, + state: FoldState::default(), + } + } + + /// Fold one item at its delivery sequence. + pub fn fold(&mut self, item: &Item<'_>, stream_seq: u64) { + match item { + Item::Platform(record) => self.fold_platform(record, stream_seq), + Item::Petri(event) => self.fold_petri(event), + } + } + + /// The run's projection, once its `run.created` record was folded. + #[must_use] + pub fn projection(&self) -> Option<&RunProjection> { + self.projection.as_ref() + } + + fn fold_petri(&mut self, event: &RunEvent) { + let at = millis(event.recorded_at); + if let Some(record) = event.coordinator() { + self.fold_coordinator(record, event, at); + } else if let Some(engine) = event.engine() { + self.fold_engine(engine, event, at); + } else if let Some(view) = event.view() { + self.fold_view(view, event, at); + } + if let Some(projection) = self.projection.as_mut() { + touch(projection, at); + } + } + + /// The shown stage an event's subject firing belongs to. + fn stage_of( + &mut self, + execution: ExecutionId, + subject: Option<&Subject>, + ) -> Option<&mut StageProjection> { + let firing = subject?.firing?; + let stage = self + .state + .stages + .get(&stage_key(execution.raw(), firing.raw()))?; + if !stage.shown { + return None; + } + let stage_id = stage.stage_id.clone(); + self.projection.as_mut()?.stage_mut(&stage_id) + } +} + +impl Default for RunView { + fn default() -> Self { + Self::new() + } +} + +// ── Shared by the folds ───────────────────────────────────────────────── + +/// Apply a status transition; one the lifecycle refuses is logged and +/// skipped, since the view never fails the run. +fn apply_status(projection: &mut RunProjection, status: RunStatus, at: DateTime) { + if let Err(error) = projection.try_apply_status(status, at) { + debug!(error = %error, "status transition not applied to the Petri projection"); + } +} + +fn touch(projection: &mut RunProjection, at: DateTime) { + if at > projection.last_event_at { + projection.last_event_at = at; + } +} + +/// The key of a stage: its execution and firing. +#[must_use] +pub fn stage_key(execution: u64, firing: u64) -> String { + format!("{execution}:{firing}") +} + +/// Which firing of its node a subject is, 1-based. +#[must_use] +pub fn visit_of(subject: &Subject) -> u32 { + subject.visit.unwrap_or(1).max(1) +} + +/// The role a frontend gave a node under `meta.kind`, or the empty string. +fn node_meta_kind(node: &NodeRef) -> &str { + node.meta.get("kind").and_then(Value::as_str).unwrap_or("") +} + +/// Whether a node is a logical stage the projection shows, or a lowering +/// node it keeps off the list: one a frontend marked synthetic, or a +/// parallel branch's delegate. +#[must_use] +pub fn is_shown(node: &NodeRef) -> bool { + let synthetic = node + .meta + .get("synthetic") + .and_then(Value::as_bool) + .unwrap_or(false); + !synthetic && node_meta_kind(node) != "parallel.branch" +} + +/// The label a shown firing takes, which is the stage id the projection +/// keys it by: `node@visit`, or `node/e@visit` when another +/// execution's firing already took that label. `taken` is every label given +/// so far; the caller adds the one returned. The interview adapter labels a +/// question's stage through this same rule, so the stage a question names +/// is the stage the projection shows. +#[must_use] +pub fn stage_label( + node_name: &str, + visit: u32, + execution: ExecutionId, + taken: &BTreeSet, +) -> StageId { + let stage_id = StageId::new(node_name.to_string(), visit); + if taken.contains(&stage_id.to_string()) { + return StageId::new(format!("{node_name}/e{}", execution.raw()), visit); + } + stage_id +} + +fn millis(recorded_at: u64) -> DateTime { + Utc.timestamp_millis_opt(i64::try_from(recorded_at).unwrap_or(i64::MAX)) + .single() + .unwrap_or_default() +} + +/// The run id a Petri run key names. +#[must_use] +pub fn run_id_of(key: &str) -> Option { + key.parse().ok() +} + +#[cfg(test)] +mod tests { + use fabro_store::platform_records::{PlatformRecord, RunCreatedRecord}; + use fabro_types::test_support as types_support; + use petri_runtime::driver::BranchRole; + use petri_runtime::ir::{FiringId, NodeId}; + + use super::*; + + #[test] + fn a_taken_label_is_made_unique_by_the_execution() { + let mut view = RunView::new(); + let created = StoredPlatformRecord { + seq: 1, + recorded_at: 1_000, + record: PlatformRecord::RunCreated(RunCreatedRecord { + spec: types_support::test_run_spec(), + title: Some("A run".to_string()), + parent_id: None, + retried_from: None, + web_url: None, + }), + position: None, + }; + view.fold(&Item::Platform(&created), 1); + let subject = |name: &str| Subject { + node: NodeRef { + id: NodeId::new(1), + name: name.into(), + kind: "attractor/command".into(), + meta: serde_json::json!({ "kind": "command" }), + }, + firing: Some(FiringId::new(4)), + visit: Some(1), + attempt: None, + generation: None, + branch: BranchRole::None, + }; + view.start_visit(ExecutionId::new(1), &subject("build"), millis(2_000)); + view.start_visit(ExecutionId::new(2), &subject("build"), millis(3_000)); + let labels: Vec = view + .projection() + .expect("the run was created") + .iter_stages() + .map(|(id, _)| id.to_string()) + .collect(); + assert_eq!(labels, vec!["build@1", "build/e2@1"]); + assert_eq!(view.state.stages.len(), 2); + } +} diff --git a/lib/components/fabro-petri/src/projection/model.rs b/lib/components/fabro-petri/src/projection/model.rs new file mode 100644 index 000000000..20d24a472 --- /dev/null +++ b/lib/components/fabro-petri/src/projection/model.rs @@ -0,0 +1,41 @@ +//! The model and usage facts the records carry, as the view names them. + +use fabro_types::ModelRef; +use lithos_llm::catalog::{ModelId, ProviderId}; +use lithos_llm::types::Usage; +use serde_json::Value; + +pub(super) fn usage_of(value: Option<&Value>) -> Option { + serde_json::from_value(value?.clone()).ok() +} + +/// `provider/model` into its parts, or the model alone. +pub(super) fn split_model(model: &str) -> (Option<&str>, &str) { + match model.split_once('/') { + Some((provider, model)) if !provider.is_empty() && !model.is_empty() => { + (Some(provider), model) + } + _ => (None, model), + } +} + +pub(super) fn model_ref(provider: Option<&str>, model: &str) -> Option { + let provider = provider.filter(|provider| !provider.is_empty())?; + Some(ModelRef::new( + ProviderId::new(provider), + ModelId::new(model), + )) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_model_selector_splits_into_provider_and_model() { + assert_eq!(split_model("openai/gpt-5.4"), (Some("openai"), "gpt-5.4")); + assert_eq!(split_model("gpt-5.4"), (None, "gpt-5.4")); + assert!(model_ref(None, "gpt-5.4").is_none()); + assert!(model_ref(Some("openai"), "gpt-5.4").is_some()); + } +} diff --git a/lib/components/fabro-petri/src/projection/platform.rs b/lib/components/fabro-petri/src/projection/platform.rs new file mode 100644 index 000000000..d5c592768 --- /dev/null +++ b/lib/components/fabro-petri/src/projection/platform.rs @@ -0,0 +1,344 @@ +//! The platform records folded into the view: the run's creation, its +//! lifecycle, its title and parent, the branch and identity the first +//! checkpoint recorded, every checkpoint and artifact, the run's diff, and +//! the pull request (VIEWS.md "Run", "Checkpoints", "Artifacts", "Pull +//! request"). + +use chrono::{DateTime, Utc}; +use fabro_store::platform_records::{ + PlatformRecord, RunLifecycleKind, RunLifecycleRecord, StoredPlatformRecord, +}; +use fabro_types::{ + CheckpointRecord as ViewCheckpoint, Conclusion, FailureCategory, FailureDetail, FailureReason, + PullRequestCreation, PullRequestCreationStatus, PullRequestLink, RunApproval, RunApprovalState, + RunArtifact, RunControlAction, RunDiff, RunFailure, RunProjection, RunSandbox, RunStatus, + RunTiming, StageOutcome, format_blob_ref, +}; +use tracing::debug; + +use super::sandbox::sandbox_plan; +use super::{RunView, apply_status, millis, stage_key, touch}; + +impl RunView { + pub(super) fn fold_platform(&mut self, stored: &StoredPlatformRecord, stream_seq: u64) { + let at = millis(stored.recorded_at); + if let PlatformRecord::RunCreated(created) = &stored.record { + let title = created + .title + .clone() + .unwrap_or_else(|| fabro_types::infer_run_title(created.spec.graph.goal())); + let mut projection = RunProjection::new(title, created.spec.clone(), at); + projection.parent_id = created.parent_id; + projection.retried_from = created.retried_from; + projection.web_url.clone_from(&created.web_url); + projection.sandbox = Some(RunSandbox::planned(sandbox_plan( + &projection.spec.settings.run.environment, + ))); + self.projection = Some(projection); + return; + } + let Some(projection) = self.projection.as_mut() else { + debug!( + seq = stored.seq, + kind = %stored.record.kind(), + "platform record before run.created; not folded" + ); + return; + }; + touch(projection, at); + match &stored.record { + PlatformRecord::RunLifecycle(record) => fold_lifecycle(projection, record, at), + PlatformRecord::RunTitle(record) => projection.title.clone_from(&record.title), + PlatformRecord::RunParent(record) => projection.parent_id = record.parent_id, + PlatformRecord::RunArchived => projection.archived_at = Some(at), + PlatformRecord::RunUnarchived => projection.archived_at = None, + PlatformRecord::RunSuperseded(record) => { + projection.superseded_by = Some(record.new_run_id); + } + PlatformRecord::RunCreated(_) + | PlatformRecord::RunNotice(_) + | PlatformRecord::InterviewAnswered(_) + | PlatformRecord::NotificationSent(_) + | PlatformRecord::RunPaired(_) => {} + PlatformRecord::RunBranch(record) => { + self.state.run_branch.clone_from(&record.run_branch); + self.state.base_sha.clone_from(&record.base_sha); + if let Some(start) = projection.start.as_mut() { + start.run_branch.clone_from(&record.run_branch); + start.base_sha.clone_from(&record.base_sha); + } + } + PlatformRecord::GitIdentity(record) => { + projection.git_identity = Some(record.identity.clone()); + } + PlatformRecord::Checkpoint(record) => { + self.state.checkpoints = self.state.checkpoints.saturating_add(1); + let stage = self + .state + .stages + .get(&stage_key(record.execution, record.firing)); + let current_node = stage.map_or_else(String::new, |stage| stage.node_name.clone()); + let stage_id = stage + .filter(|stage| stage.shown) + .map(|stage| stage.stage_id.clone()); + let checkpoint = fabro_types::Checkpoint { + timestamp: at, + current_node: current_node.clone(), + git_commit_sha: record.git_commit_sha.clone(), + }; + // The patch stays in the blob table; the view carries its + // reference for a reader to resolve. + let patch = record.patch_blob.as_ref().map(format_blob_ref); + if let Some(stage) = stage_id.and_then(|stage_id| projection.stage_mut(&stage_id)) { + if patch.is_some() { + stage.diff.clone_from(&patch); + } + } + projection.checkpoints.push(ViewCheckpoint { + seq: u32::try_from(stream_seq).unwrap_or(u32::MAX), + checkpoint, + diff: RunDiff { + patch, + summary: record.diff_summary, + }, + }); + } + PlatformRecord::ArtifactCollected(record) => { + let stage = self + .state + .stages + .get(&stage_key(record.execution, record.firing)); + let Some(stage_id) = stage.map(|stage| stage.stage_id.clone()) else { + debug!( + seq = stored.seq, + path = record.path, + "artifact record for an unknown firing; not folded" + ); + return; + }; + projection.artifacts.push(RunArtifact { + stage_id, + retry: record.attempt, + relative_path: record.path.clone(), + size: record.bytes, + blob: record.blob, + }); + } + PlatformRecord::RunDiff(record) => { + let diff = RunDiff { + patch: record.patch_blob.as_ref().map(format_blob_ref), + summary: record.diff_summary, + }; + if let Some(conclusion) = projection.conclusion.as_mut() { + conclusion.diff = diff.clone(); + } + self.state.run_diff = Some(diff); + } + PlatformRecord::PullRequestRequested(record) => { + projection.pull_request_creation = Some(PullRequestCreation { + id: record.creation_id, + status: PullRequestCreationStatus::Pending, + model: record.model.clone(), + force: record.force, + requested_at: at, + updated_at: at, + pull_request: None, + error: None, + }); + } + PlatformRecord::PullRequestCreated(record) => { + let link = PullRequestLink { + owner: record.owner.clone(), + repo: record.repo.clone(), + number: record.number, + }; + projection.pull_request = Some(link.clone()); + if let Some(creation) = projection + .pull_request_creation + .as_mut() + .filter(|creation| creation.is_pending()) + { + creation.succeed(link, at); + } + } + PlatformRecord::PullRequestFailed(record) => { + if let Some(creation) = + projection + .pull_request_creation + .as_mut() + .filter(|creation| { + creation.is_pending() + && record + .creation_id + .is_none_or(|creation_id| creation_id == creation.id) + }) + { + creation.fail(record.error.clone(), at); + } + } + PlatformRecord::PullRequestLinked(record) => { + let link = record.link(); + projection.pull_request = Some(link.clone()); + if let Some(creation) = projection + .pull_request_creation + .as_mut() + .filter(|creation| creation.is_pending()) + { + creation.succeed(link, at); + } + } + PlatformRecord::PullRequestUnlinked(_) => { + projection.pull_request = None; + projection.pull_request_creation = None; + } + } + } +} + +fn fold_lifecycle(projection: &mut RunProjection, record: &RunLifecycleRecord, at: DateTime) { + use RunLifecycleKind as Kind; + match record.transition { + Kind::Submitted => apply_status(projection, RunStatus::Submitted, at), + Kind::StartRequested => {} + Kind::Unpaused => { + let status = match projection.status { + RunStatus::Paused { + prior_block: Some(blocked_reason), + } => RunStatus::Blocked { blocked_reason }, + _ => RunStatus::Running, + }; + apply_status(projection, status, at); + if projection.pending_control == Some(RunControlAction::Unpause) { + projection.pending_control = None; + } + } + Kind::Pending => { + if let Some(status) = record.status { + apply_status(projection, status, at); + } + projection.approval = Some(RunApproval { + state: RunApprovalState::Pending, + requested_at: at, + decided_at: None, + denial_reason: None, + }); + } + Kind::Approved => { + if let Some(approval) = projection.approval.as_mut() { + approval.state = RunApprovalState::Approved; + approval.decided_at = Some(at); + } + } + Kind::Denied => { + if let Some(approval) = projection.approval.as_mut() { + approval.state = RunApprovalState::Denied; + approval.decided_at = Some(at); + approval.denial_reason.clone_from(&record.reason); + } + apply_status( + projection, + RunStatus::Failed { + reason: FailureReason::ApprovalDenied, + }, + at, + ); + } + Kind::Runnable => { + // A run left in flight by a restart goes back to the queue: the + // resume's `runnable` steps back from wherever the run stood. + let in_flight = matches!( + projection.status, + RunStatus::Starting + | RunStatus::Running + | RunStatus::Blocked { .. } + | RunStatus::Paused { .. } + ); + if in_flight && record.status == Some(RunStatus::Runnable) { + projection.status = RunStatus::Runnable; + projection.status_updated_at = at; + } else if let Some(status) = record.status { + apply_status(projection, status, at); + } + } + Kind::Blocked => { + // A block that lands while the run is paused waits behind the + // pause: the unpause restores it. + match (projection.status, record.status) { + (RunStatus::Paused { .. }, Some(RunStatus::Blocked { blocked_reason })) => { + apply_status( + projection, + RunStatus::Paused { + prior_block: Some(blocked_reason), + }, + at, + ); + } + (_, Some(status)) => apply_status(projection, status, at), + (_, None) => {} + } + } + Kind::Unblocked => { + let status = match projection.status { + RunStatus::Paused { .. } => RunStatus::Paused { prior_block: None }, + _ => record.status.unwrap_or(RunStatus::Running), + }; + apply_status(projection, status, at); + } + Kind::Starting | Kind::Running | Kind::Removing | Kind::Dead => { + if let Some(status) = record.status { + apply_status(projection, status, at); + } + } + Kind::Paused => { + let prior_block = match projection.status { + RunStatus::Blocked { blocked_reason } => Some(blocked_reason), + RunStatus::Paused { prior_block } => prior_block, + _ => None, + }; + apply_status(projection, RunStatus::Paused { prior_block }, at); + if projection.pending_control == Some(RunControlAction::Pause) { + projection.pending_control = None; + } + } + Kind::Succeeded | Kind::Failed => { + if let Some(status) = record.status { + apply_status(projection, status, at); + } + projection.pending_control = None; + if projection.conclusion.is_none() { + let (outcome, failure) = match record.status { + Some(RunStatus::Failed { reason }) => ( + StageOutcome::Failed { + retry_requested: false, + }, + Some(RunFailure { + reason, + detail: FailureDetail::new( + record + .reason + .clone() + .unwrap_or_else(|| "the run failed".to_string()), + FailureCategory::Deterministic, + ), + }), + ), + _ => (StageOutcome::Succeeded, None), + }; + projection.conclusion = Some(Conclusion { + timestamp: at, + status: outcome, + timing: RunTiming::default(), + failure, + final_git_commit_sha: None, + stages: Vec::new(), + usage: None, + total_retries: 0, + diff: RunDiff::default(), + }); + } + } + Kind::CancelRequested => projection.pending_control = Some(RunControlAction::Cancel), + Kind::PauseRequested => projection.pending_control = Some(RunControlAction::Pause), + Kind::UnpauseRequested => projection.pending_control = Some(RunControlAction::Unpause), + } +} diff --git a/lib/components/fabro-petri/src/projection/progress.rs b/lib/components/fabro-petri/src/projection/progress.rs new file mode 100644 index 000000000..7bfe03428 --- /dev/null +++ b/lib/components/fabro-petri/src/projection/progress.rs @@ -0,0 +1,423 @@ +//! A step's progress records folded into its stage: a command's log lines, +//! the payloads the Attractor steps emit (the prompt and its completion, +//! the fallback plan, the tools a session was offered, a parallel branch's +//! start), Pebble's coding-agent envelope, and a gate's question (VIEWS.md +//! "Agent activity", "Questions"). + +use chrono::{DateTime, Utc}; +use fabro_types::{ + BlockedReason, CodingAgentEvent, CodingEvent, InterviewOption, InterviewQuestionRecord, + PendingInterviewRecord, ReviewTarget, ReviewTargetKind, RunStatus, StageInferenceProjection, + StageModelUsage, StageProjection, ToolCategory, ToolSource, ToolSummary, timing, +}; +use petri_execution::ExecutionId; +use petri_execution::events::{Parsed, RunEvent}; +use petri_runtime::ir::StepEvent; +use petri_runtime::steps::QuestionReference; +use serde_json::Value; +use tracing::debug; + +use super::model::{model_ref, split_model, usage_of}; +use super::{RunView, apply_status, stage_key}; +use crate::interview::question_type; + +impl RunView { + pub(super) fn fold_progress( + &mut self, + execution: ExecutionId, + event: &RunEvent, + ev: &StepEvent, + at: DateTime, + ) { + match ev { + StepEvent::Log { line, .. } => { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + let output = stage.output.get_or_insert_default(); + output.push_str(line); + output.push('\n'); + stage.output_bytes = Some(output.len() as u64); + stage.live_streaming = Some(true); + } + } + StepEvent::Artifact { .. } => {} + StepEvent::Custom(payload) => { + if let Some(parsed) = event.parsed() { + self.fold_parsed(execution, event, parsed, at); + return; + } + let kind = payload.get("kind").and_then(Value::as_str).unwrap_or(""); + match kind { + "pebble" => self.fold_pebble(execution, event, payload, at), + "attractor.prompt" => { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + stage.prompt = payload + .get("prompt") + .and_then(Value::as_str) + .map(str::to_string); + let model = payload.get("model").and_then(Value::as_str); + if let Some(model) = model { + let (provider, model_id) = split_model(model); + stage.provider_used = Some(StageModelUsage { + mode: StageModelUsage::MODE_PROMPT.to_string(), + provider: provider.map(str::to_string), + model: Some(model_id.to_string()), + reasoning_effort: None, + speed: None, + }); + stage.model = model_ref(provider, model_id); + } + } + } + "attractor.prompt.completed" => { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + stage.response = payload + .get("response") + .and_then(Value::as_str) + .map(str::to_string); + if let Some(usage) = usage_of(payload.get("usage")) { + stage.usage = usage; + } + } + } + "attractor.fallback.plan" => { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + let route = payload + .get("routes") + .and_then(Value::as_array) + .and_then(|routes| routes.first()); + if let Some(route) = route { + let provider = route.get("provider").and_then(Value::as_str); + let model = route.get("model").and_then(Value::as_str); + stage.provider_used = Some(StageModelUsage { + mode: StageModelUsage::MODE_AGENT.to_string(), + provider: provider.map(str::to_string), + model: model.map(str::to_string), + reasoning_effort: None, + speed: None, + }); + if let Some(model) = model { + stage.model = model_ref(provider, model); + } + } + } + } + // The tools a native session was offered, once per + // session (VIEWS.md "Agent activity", tools available): + // the stage's list is the union over its sessions, by + // name, in the order the sessions listed them. + "attractor.tools" => { + if let Some(stage) = self.stage_of(execution, event.subject.as_ref()) { + let tools = payload.get("tools").and_then(Value::as_array); + for tool in tools.into_iter().flatten() { + let Some(summary) = tool_summary(tool) else { + continue; + }; + if !stage + .agent_tools + .iter() + .any(|known| known.name == summary.name) + { + stage.agent_tools.push(summary); + } + } + } + } + "attractor.parallel.branch.started" => { + let invocation = payload.get("invocation").and_then(Value::as_u64); + let index = payload + .get("index") + .and_then(Value::as_u64) + .and_then(|index| u32::try_from(index).ok()); + let fork_firing = payload + .get("occurrence") + .and_then(|occurrence| occurrence.get("firing")) + .and_then(Value::as_u64); + if let (Some(invocation), Some(index), Some(fork_firing)) = + (invocation, index, fork_firing) + { + let group = self + .state + .stages + .get(&stage_key(execution.raw(), fork_firing)) + .map(|stage| stage.stage_id.clone()); + if let Some(group) = group { + self.state.invocations.entry(invocation).or_default().branch = + Some((group, index)); + } + } + } + _ => {} + } + } + } + } + + fn fold_parsed( + &mut self, + execution: ExecutionId, + event: &RunEvent, + parsed: &Parsed, + at: DateTime, + ) { + match parsed { + Parsed::Question { question } => { + let Some(subject) = event.subject.as_ref() else { + return; + }; + let Some(firing) = subject.firing else { + return; + }; + let key = stage_key(execution.raw(), firing.raw()); + let label = self.state.stages.get(&key).map_or_else( + || subject.node.name.to_string(), + |stage| stage.stage_id.to_string(), + ); + self.state.questions.insert(question.id.clone(), key); + let Some(projection) = self.projection.as_mut() else { + return; + }; + projection + .pending_interviews + .insert(question.id.clone(), PendingInterviewRecord { + question: InterviewQuestionRecord { + id: question.id.clone(), + text: question.text.clone(), + stage: label, + question_type: question_type(question), + options: question + .options + .iter() + .map(|option| InterviewOption { + key: option.key.clone(), + label: option.label.clone(), + description: option.description.clone(), + preview: option.preview.clone(), + }) + .collect(), + allow_freeform: question.freeform, + timeout_seconds: question + .timeout_ms + .map(|timeout| timeout as f64 / 1000.0), + context_display: question.context.clone(), + review_target: question.reference.as_ref().and_then(review_target), + }, + started_at: at, + }); + apply_status( + projection, + RunStatus::Blocked { + blocked_reason: BlockedReason::HumanInputRequired, + }, + at, + ); + } + Parsed::QuestionExpired { expired } => { + self.close_questions(Some(expired.question.as_str()), None, at); + } + Parsed::Note { .. } => {} + } + } + + /// Close one question by id, or every question of a firing, and unblock + + /// Close one question by id, or every question of a firing, and unblock + /// the run when none is left. + pub(super) fn close_questions( + &mut self, + question: Option<&str>, + firing_key: Option<&str>, + at: DateTime, + ) { + let closed: Vec = match (question, firing_key) { + (Some(question), _) => vec![question.to_string()], + (None, Some(key)) => self + .state + .questions + .iter() + .filter(|(_, asked_by)| asked_by.as_str() == key) + .map(|(id, _)| id.clone()) + .collect(), + (None, None) => Vec::new(), + }; + for id in &closed { + self.state.questions.remove(id); + } + let Some(projection) = self.projection.as_mut() else { + return; + }; + for id in &closed { + projection.pending_interviews.remove(id); + } + if projection.pending_interviews.is_empty() + && matches!(projection.status, RunStatus::Blocked { .. }) + { + apply_status(projection, RunStatus::Running, at); + } + } + + fn fold_pebble( + &mut self, + execution: ExecutionId, + event: &RunEvent, + payload: &Value, + at: DateTime, + ) { + let Some(envelope) = payload.get("event") else { + return; + }; + let envelope: CodingAgentEvent = match serde_json::from_value(envelope.clone()) { + Ok(envelope) => envelope, + Err(error) => { + debug!(error = %error, "a pebble envelope did not decode; skipped"); + return; + } + }; + let Some(stage) = self.stage_of(execution, event.subject.as_ref()) else { + return; + }; + let agent = stage.agent.get_or_insert_default(); + agent.apply(&envelope); + if stage.completion.is_none() { + stage.usage = agent.usage.saturating_add(agent.descendant_usage()); + } + // A tool the stage's list names was called, by any of its sessions. + if let CodingEvent::ToolCallStarted { tool_name, .. } = &envelope.event { + if let Some(tool) = stage + .agent_tools + .iter_mut() + .find(|tool| tool.name == *tool_name) + { + tool.invoked = true; + } + } + let is_root = envelope.parent_session_id.is_none(); + #[expect( + clippy::wildcard_enum_match_arm, + reason = "pebble's event vocabulary is non-exhaustive and only some events project" + )] + match &envelope.event { + CodingEvent::SessionStarted { + provider, model, .. + } if is_root => { + stage.provider_used = Some(StageModelUsage { + mode: StageModelUsage::MODE_AGENT.to_string(), + provider: provider.clone(), + model: model.clone(), + reasoning_effort: None, + speed: None, + }); + if let Some(model) = model.as_deref() { + stage.model = model_ref(provider.as_deref(), model); + } + } + CodingEvent::LlmRequestStarted { requested_model } if is_root => { + stage.inference = Some(StageInferenceProjection { + session_id: envelope.session_id.clone(), + started_at: at, + requested_model: requested_model.clone(), + first_output_at: None, + first_output_kind: None, + retries: 0, + }); + } + CodingEvent::LlmFirstOutput { kind } => { + if let Some(inference) = stage.inference.as_mut() { + if inference.session_id == envelope.session_id { + inference.first_output_at = Some(at); + inference.first_output_kind = Some(*kind); + } + } + } + CodingEvent::LlmRetry { .. } => { + if let Some(inference) = stage.inference.as_mut() { + if inference.session_id == envelope.session_id { + inference.retries = inference.retries.saturating_add(1); + inference.first_output_at = None; + inference.first_output_kind = None; + } + } + } + CodingEvent::AssistantMessage { model, .. } => { + if is_root { + if let Some(provider) = stage + .provider_used + .as_ref() + .and_then(|used| used.provider.as_deref()) + { + stage.model = model_ref(Some(provider), model); + } + } + close_inference(stage, &envelope.session_id, at); + } + CodingEvent::Error { .. } | CodingEvent::RoundInterrupted { .. } => { + close_inference(stage, &envelope.session_id, at); + } + CodingEvent::SessionEnded => { + close_inference(stage, &envelope.session_id, at); + stage.close_tool_batch_for_session(&envelope.session_id, at); + } + CodingEvent::ToolCallStarted { tool_call_id, .. } if is_root => { + stage.open_tool_call(envelope.session_id.clone(), tool_call_id.clone(), at); + } + CodingEvent::ToolCallCompleted { tool_call_id, .. } if is_root => { + stage.close_tool_call(&envelope.session_id, tool_call_id, at); + } + _ => {} + } + } +} + +/// One tool of an `attractor.tools` payload as the stage's list carries +/// it: the name and description as recorded, Pebble's `source` as it is, +/// and Pebble's behavioural category where Petri's says which (a +/// sub-agent tool); every other tool is `other`, because the payload +/// carries Petri's origin category (`builtin`, `mcp`, `host`, `question`), +/// not Pebble's permission class. `invoked` starts false and flips on the +/// session's `ToolCallStarted`. +fn tool_summary(tool: &Value) -> Option { + let name = tool.get("name").and_then(Value::as_str)?; + let description = tool + .get("description") + .and_then(Value::as_str) + .unwrap_or_default(); + let source = tool + .get("source") + .cloned() + .and_then(|source| serde_json::from_value::(source).ok()) + .unwrap_or_default(); + let category = match tool.get("category").and_then(Value::as_str) { + Some("subagent") => ToolCategory::Subagent, + _ => ToolCategory::Other, + }; + Some(ToolSummary { + name: name.to_string(), + description: description.to_string(), + source, + category, + invoked: false, + }) +} + +/// The question's `reference` as Fabro's review target, when it is one +/// Fabro's validation admits (a `document`, or a reference without a kind, +/// with a label and an absolute HTTP URL within Fabro's limits). +fn review_target(reference: &QuestionReference) -> Option { + let kind = match reference.kind.as_deref() { + Some("document") | None => ReviewTargetKind::Document, + Some(_) => return None, + }; + ReviewTarget::new(reference.label.clone(), reference.url.clone(), kind).ok() +} + +fn close_inference(stage: &mut StageProjection, session_id: &str, at: DateTime) { + let open = stage + .inference + .as_ref() + .is_some_and(|inference| inference.session_id == session_id); + if !open { + return; + } + if let Some(inference) = stage.inference.take() { + stage.accumulate_inference_ms(timing::elapsed_ms(inference.started_at, at)); + } +} diff --git a/lib/components/fabro-petri/src/projection/sandbox.rs b/lib/components/fabro-petri/src/projection/sandbox.rs new file mode 100644 index 000000000..61ee6dca9 --- /dev/null +++ b/lib/components/fabro-petri/src/projection/sandbox.rs @@ -0,0 +1,76 @@ +//! The run's sandbox as the view shows it: the plan its settings give, and +//! the instance Petri's scope records name (VIEWS.md "Sandbox"). + +use fabro_types::settings::run::RunEnvironmentSettings; +use fabro_types::{ + RunProjection, RunSandboxInstance, RunSandboxPlan, RunSandboxRuntime, SandboxProviderKind, +}; +use petri_runtime::ir::SandboxInstance; + +pub(super) fn sandbox_plan(settings: &RunEnvironmentSettings) -> RunSandboxPlan { + RunSandboxPlan { + provider: settings.provider.clone(), + image: (settings.provider == SandboxProviderKind::DOCKER) + .then(|| settings.image.docker.clone()) + .flatten() + .filter(|image| !image.is_empty()), + snapshot: None, + } +} + +/// The plan the projection's sandbox carries, or the one its environment +/// settings give when no sandbox was projected yet. +pub(super) fn sandbox_plan_of(projection: &RunProjection) -> RunSandboxPlan { + projection.sandbox.as_ref().map_or_else( + || sandbox_plan(&projection.spec.settings.run.environment), + |sandbox| sandbox.plan().clone(), + ) +} + +/// Fabro's name for the provider Petri's `scope.acquired` names: Petri's +/// `host` is Fabro's `local`; every other kind is spelled the same. `None` +/// for a name that is no provider kind. +pub(super) fn provider_kind(provider: &str) -> Option { + if provider == "host" { + return Some(SandboxProviderKind::LOCAL); + } + SandboxProviderKind::try_new(provider).ok() +} + +/// The run's sandbox instance from Petri's record of the scope's +/// acquisition: the provider, the provider's id for the sandbox (what a +/// reconnect attaches by), its image and snapshot when the provider knows +/// them, the working directory, and how long the acquisition took. The +/// clone fields stay unset: Petri's checkout copies the bound repository +/// into the workspace and is not a clone Fabro made, and the workspace +/// roots are the provider's own layout, read live. `retained` waits for +/// the scope's release. +pub(super) fn sandbox_instance( + plan: &RunSandboxPlan, + sandbox: &SandboxInstance, + ready_duration_ms: u64, +) -> RunSandboxInstance { + RunSandboxInstance { + provider: provider_kind(&sandbox.provider) + .unwrap_or_else(|| plan.provider.clone()), + image: sandbox + .image + .as_ref() + .map(ToString::to_string) + .or_else(|| plan.image.clone()), + snapshot: sandbox.snapshot.as_ref().map(ToString::to_string), + runtime: RunSandboxRuntime { + id: sandbox.instance.to_string(), + working_directory: sandbox.working_directory.to_string(), + repo_cloned: None, + clone_origin_url: None, + clone_branch: None, + workspace_root: None, + repos_root: None, + primary_repo_path: None, + primary_repo_link: None, + }, + ready_duration_ms: Some(ready_duration_ms), + retained: None, + } +}