Ingest runtime contract events into storage
This commit is contained in:
1 parent
98f975a4ca
commit
1959d218d9
5 files changed
+144
-1
No files matched your search
@@ -8,6 +8,7 @@ version.workspace = true
|
||||
[dependencies]
|
||||
anyhow.workspace = true
|
||||
domain.workspace = true
|
||||
pi_bridge.workspace = true
|
||||
rusqlite.workspace = true
|
||||
serde_json.workspace = true
|
||||
support.workspace = true
|
||||
|
||||
@@ -4,6 +4,7 @@ use domain::{
|
||||
Activity, Agent, AgentMessage, Approval, ApprovalState, ArtifactRecord, Company, Connector,
|
||||
Document, Goal, ProfileSnapshot, Project, Run, RunEventRecord, Task, Workspace,
|
||||
};
|
||||
use pi_bridge::RuntimeEvent;
|
||||
use rusqlite::{Connection, OptionalExtension, params};
|
||||
use std::{
|
||||
env,
|
||||
@@ -270,6 +271,84 @@ impl ProfileStore {
|
||||
Ok(artifact_id)
|
||||
}
|
||||
|
||||
pub fn ingest_runtime_event(&mut self, event: RuntimeEvent) -> Result<()> {
|
||||
match event {
|
||||
RuntimeEvent::HandshakeAck { .. } => Ok(()),
|
||||
RuntimeEvent::Message { run_id, role, body } => {
|
||||
let company_id = self.company_id_for_run(&run_id)?;
|
||||
let title = format!("message:{role}");
|
||||
self.append_run_event(&company_id, &run_id, "message", &title, &body)?;
|
||||
Ok(())
|
||||
}
|
||||
RuntimeEvent::ToolCall {
|
||||
run_id,
|
||||
tool_call_id,
|
||||
tool_name,
|
||||
input_json,
|
||||
} => {
|
||||
let company_id = self.company_id_for_run(&run_id)?;
|
||||
let detail = serde_json::to_string(&input_json)
|
||||
.context("failed to encode tool call payload")?;
|
||||
let title = format!("{tool_name}:{tool_call_id}");
|
||||
self.append_run_event(&company_id, &run_id, "tool_call", &title, &detail)?;
|
||||
Ok(())
|
||||
}
|
||||
RuntimeEvent::ToolResult {
|
||||
run_id,
|
||||
tool_call_id,
|
||||
output_json,
|
||||
} => {
|
||||
let company_id = self.company_id_for_run(&run_id)?;
|
||||
let detail = serde_json::to_string(&output_json)
|
||||
.context("failed to encode tool result payload")?;
|
||||
let title = format!("tool_result:{tool_call_id}");
|
||||
self.append_run_event(&company_id, &run_id, "tool_result", &title, &detail)?;
|
||||
Ok(())
|
||||
}
|
||||
RuntimeEvent::Artifact {
|
||||
run_id,
|
||||
provider,
|
||||
external_id,
|
||||
title,
|
||||
metadata_json,
|
||||
} => {
|
||||
let company_id = self.company_id_for_run(&run_id)?;
|
||||
self.register_artifact(
|
||||
&company_id,
|
||||
&run_id,
|
||||
&provider,
|
||||
&external_id,
|
||||
&title,
|
||||
&metadata_json.to_string(),
|
||||
)?;
|
||||
self.append_run_event(&company_id, &run_id, "artifact", &title, &external_id)?;
|
||||
Ok(())
|
||||
}
|
||||
RuntimeEvent::RunFinished {
|
||||
run_id,
|
||||
status,
|
||||
summary,
|
||||
session_id: _,
|
||||
cost_usd,
|
||||
} => {
|
||||
let company_id = self.company_id_for_run(&run_id)?;
|
||||
let ts = now_iso()?;
|
||||
self.conn.execute(
|
||||
"update runs set status = ?1, summary = ?2, cost_usd = ?3, finished_at = ?4, updated_at = ?4 where id = ?5",
|
||||
params![status, summary, cost_usd, ts, run_id],
|
||||
)?;
|
||||
self.append_run_event(
|
||||
&company_id,
|
||||
&run_id,
|
||||
"run_finished",
|
||||
"run_finished",
|
||||
&summary,
|
||||
)?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn decide_approval(
|
||||
&mut self,
|
||||
company_id: &str,
|
||||
@@ -828,6 +907,17 @@ impl ProfileStore {
|
||||
None => Ok(Vec::new()),
|
||||
}
|
||||
}
|
||||
|
||||
fn company_id_for_run(&self, run_id: &str) -> Result<String> {
|
||||
self.conn
|
||||
.query_row(
|
||||
"select company_id from runs where id = ?1",
|
||||
params![run_id],
|
||||
|row| row.get(0),
|
||||
)
|
||||
.optional()?
|
||||
.ok_or_else(|| anyhow!("run {run_id} was not found"))
|
||||
}
|
||||
}
|
||||
|
||||
fn query_many<T, P, F>(conn: &Connection, sql: &str, params: P, mut mapper: F) -> Result<Vec<T>>
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
use pi_bridge::RuntimeEvent;
|
||||
use storage::{NewCompanyInput, ProfileStore};
|
||||
use tempfile::tempdir;
|
||||
|
||||
@@ -44,6 +45,32 @@ fn runtime_contract_entities_round_trip_through_snapshot() {
|
||||
)
|
||||
.expect("register artifact");
|
||||
|
||||
store
|
||||
.ingest_runtime_event(RuntimeEvent::Message {
|
||||
run_id: run_id.clone(),
|
||||
role: "assistant".to_string(),
|
||||
body: "Draft ready".to_string(),
|
||||
})
|
||||
.expect("ingest message event");
|
||||
store
|
||||
.ingest_runtime_event(RuntimeEvent::Artifact {
|
||||
run_id: run_id.clone(),
|
||||
provider: "filesystem".to_string(),
|
||||
external_id: "/tmp/orbit-runtime/output.txt".to_string(),
|
||||
title: "output.txt".to_string(),
|
||||
metadata_json: serde_json::json!({"path": "/tmp/orbit-runtime/output.txt"}),
|
||||
})
|
||||
.expect("ingest artifact event");
|
||||
store
|
||||
.ingest_runtime_event(RuntimeEvent::RunFinished {
|
||||
run_id: run_id.clone(),
|
||||
status: "completed".to_string(),
|
||||
summary: "Run complete".to_string(),
|
||||
session_id: Some("session-1".to_string()),
|
||||
cost_usd: Some(1.25),
|
||||
})
|
||||
.expect("ingest run finished event");
|
||||
|
||||
let snapshot = store.snapshot().expect("snapshot");
|
||||
assert!(snapshot.run_events.iter().any(|event| {
|
||||
event.id == event_id
|
||||
@@ -57,6 +84,24 @@ fn runtime_contract_entities_round_trip_through_snapshot() {
|
||||
&& artifact.provider == "filesystem"
|
||||
&& artifact.title == "README update"
|
||||
}));
|
||||
assert!(snapshot.run_events.iter().any(|event| {
|
||||
event.run_id == run_id && event.event_kind == "message" && event.detail == "Draft ready"
|
||||
}));
|
||||
assert!(
|
||||
snapshot
|
||||
.artifacts
|
||||
.iter()
|
||||
.any(|artifact| { artifact.run_id == run_id && artifact.title == "output.txt" })
|
||||
);
|
||||
let run = snapshot
|
||||
.runs
|
||||
.iter()
|
||||
.find(|run| run.id == run_id)
|
||||
.expect("run present");
|
||||
assert_eq!(run.status, "completed");
|
||||
assert_eq!(run.summary, "Run complete");
|
||||
assert_eq!(run.cost_usd, Some(1.25));
|
||||
assert!(run.finished_at.is_some());
|
||||
|
||||
let reopened = ProfileStore::open(sandbox.path()).expect("reopen store");
|
||||
let reopened_snapshot = reopened.snapshot().expect("reopened snapshot");
|
||||
@@ -72,4 +117,10 @@ fn runtime_contract_entities_round_trip_through_snapshot() {
|
||||
.iter()
|
||||
.any(|artifact| artifact.id == artifact_id)
|
||||
);
|
||||
assert!(
|
||||
reopened_snapshot
|
||||
.runs
|
||||
.iter()
|
||||
.any(|run| run.id == run_id && run.status == "completed")
|
||||
);
|
||||
}
|
||||
@@ -114,7 +114,7 @@
|
||||
],
|
||||
"priority": 8,
|
||||
"passes": false,
|
||||
"notes": "Core completion story for full replacement."
|
||||
"notes": "In progress: runtime events and artifacts now persist into SQLite, surface in ProfileSnapshot, render in Runs UI, and pass storage-backed e2e verification."
|
||||
},
|
||||
{
|
||||
"id": "US-009",
|
||||
|
||||
@@ -12,3 +12,4 @@ Started: 2026-04-19
|
||||
- Completed first native motion/state parity layer using GPUI animation wrappers for onboarding, inbox, and runs.
|
||||
- Added runtime contract persistence entities for run events and artifacts, exposed them in ProfileSnapshot, and rendered them in the native runs inspector.
|
||||
- Added runtime_contract_snapshot_e2e.rs to verify runtime event and artifact persistence across reopen.
|
||||
- Runtime contract integration advanced: RuntimeEvent ingestion now persists message/tool/artifact/run-finished events into storage and exposes them in the runs inspector.
|
||||
Reference in new issue
Block a user