From b893a72256afb129bc198c3e185b09f43afb5538 Mon Sep 17 00:00:00 2001 From: "liqiankun.1111" Date: Sun, 9 Aug 2026 16:03:45 +0800 Subject: [PATCH] feat: support orchestrated agent sessions --- README.md | 37 +++++++++++--- go/recorder.go | 48 ++++++++++++++++-- go/store_test.go | 41 +++++++++++++++ go/types.go | 5 ++ python/examples/orchestrated_agents.py | 47 ++++++++++++++++++ python/tests/test_models.py | 21 ++++++++ python/tests/test_recorder.py | 37 +++++++++----- spec/rfcs/0001-agent-ledger.md | 58 +++++++++++++++++++--- spec/rfcs/0002-polyglot-adapters.md | 4 ++ spec/schemas/event.schema.json | 4 ++ typescript/packages/core/src/recorder.ts | 40 +++++++++++++++ typescript/packages/core/test/core.test.ts | 39 ++++++++++++++- typescript/packages/pi/src/harness.ts | 5 +- 13 files changed, 357 insertions(+), 29 deletions(-) create mode 100644 python/examples/orchestrated_agents.py diff --git a/README.md b/README.md index 46b598d..045794a 100644 --- a/README.md +++ b/README.md @@ -1,21 +1,40 @@ # Agent Ledger Agent Ledger is a framework-neutral specification and a set of polyglot adapters for durable agent -sessions. It records model and tool attempts before execution, preserves causal timelines across -distributed agents, and lets each framework rebuild its own native session after interruption. +execution records. Agent loops and orchestrators append to the same session history, producing a +causal account of model calls, tool calls, delegation, framework-native state, and outcomes. The specification is the stable product. Language SDKs are deliberately small; most project code lives in adapters that understand a framework's hooks, messages, checkpoints, and resume APIs. +## Architecture position + +Agent Ledger is not another agent loop or workflow engine. It is a shared evidence layer across +both: + +```text +Orchestrator ── decisions, delegation, approvals ──┐ + ├── Agent Ledger session +Agent loops ── steps, attempts, native state ──────┘ │ + ├── recovery + ├── global timeline + └── analysis / evaluation +``` + +An orchestrator owns desired state, scheduling, and run ownership. Each agent framework owns its +native context and resume API. Agent Ledger owns the immutable facts that let those systems explain +and reconstruct what happened. + ## Model -- `Session` groups one end-to-end task across processes, languages, and agents. -- `Run` identifies one semantic agent execution and participates in the causal DAG. +- `Session` groups one end-to-end task across processes, languages, agents, and orchestration runs. +- `Run` identifies one semantic execution by an agent or orchestrator and participates in the + causal DAG. - `EventStream` is an optimistic-concurrency partition. It may contain one run's execution events or framework-native state that survives several runtime runs. - `Step` is logical work that survives retries; `Attempt` is one physical model or tool invocation. - Normalized events are the source for timelines and trajectories. Framework-native records are the - source for resume. + lossless input to framework-owned resume. Requested events are committed before an external call. A requested event without a terminal event is unresolved after a crash. It is input to the adapter's reconciliation policy; an adapter must @@ -31,7 +50,7 @@ not silently replay a side-effecting tool. | `typescript/` | TypeScript core SDK and Pi adapter | | `go/` | Go core SDK and AgentGo adapter | -Current framework profiles: +Current framework profiles are integration examples, not definitions of the core session model: | Adapter | Recording | Recovery | | --- | --- | --- | @@ -42,6 +61,10 @@ Current framework profiles: Every adapter publishes machine-readable capabilities such as `strict`, `best_effort`, and `unsupported`; installing a telemetry-only hook never silently claims durable recovery. +Pi's append-only session tree is preserved in a dedicated framework stream because Pi needs it for +lossless reconstruction. Its entry types, active leaf, and branching rules remain Pi-owned rather +than becoming requirements for other agents or orchestrators. + ## Store contract Applications inject an `EventStore`. V1 has no mandatory collector or `/agent-session` service: @@ -69,3 +92,5 @@ make build See [RFC 0001](spec/rfcs/0001-agent-ledger.md) for the ledger contract and [RFC 0002](spec/rfcs/0002-polyglot-adapters.md) for framework recording and recovery boundaries. +The [orchestrated agents example](python/examples/orchestrated_agents.py) shows an orchestrator and +multiple agent loops contributing to one causal session. diff --git a/go/recorder.go b/go/recorder.go index ee29018..2d4f3b8 100644 --- a/go/recorder.go +++ b/go/recorder.go @@ -13,6 +13,7 @@ type RecorderOptions struct { RunID string StreamID string Actor Actor + Parent *CausalParent ExpectedVersion *int64 } @@ -21,6 +22,7 @@ type SessionRecorder struct { stream EventStream runID string actor Actor + parent *CausalParent expectedVersion int64 mu sync.Mutex } @@ -34,11 +36,17 @@ func NewSessionRecorder(options RecorderOptions) *SessionRecorder { if options.ExpectedVersion != nil { expectedVersion = *options.ExpectedVersion } + var parent *CausalParent + if options.Parent != nil { + copy := *options.Parent + parent = © + } return &SessionRecorder{ store: options.Store, stream: EventStream{SessionID: options.SessionID, StreamID: streamID}, runID: options.RunID, actor: options.Actor, + parent: parent, expectedVersion: expectedVersion, } } @@ -64,12 +72,39 @@ func (r *SessionRecorder) RunID() string { return r.runID } func (r *SessionRecorder) Store() EventStore { return r.store } func (r *SessionRecorder) Record(ctx context.Context, eventType string, payload map[string]any, stepID, attemptID string) (StoredEvent, error) { - r.mu.Lock() - defer r.mu.Unlock() event := NewEvent(eventType, r.stream.SessionID, r.runID, r.actor) - event.Payload = payload + event.Payload = payloadOrEmpty(payload) event.StepID = stepID event.AttemptID = attemptID + return r.appendEvent(ctx, event) +} + +func (r *SessionRecorder) StartRun(ctx context.Context, payload map[string]any) (StoredEvent, error) { + event := NewEvent("run.started", r.stream.SessionID, r.runID, r.actor) + event.Payload = payloadOrEmpty(payload) + if r.parent != nil { + event.ParentRunID = r.parent.RunID + event.CausedByEventID = r.parent.CausedByEventID + } + return r.appendEvent(ctx, event) +} + +func (r *SessionRecorder) Child(runID string, actor Actor, causedByEventID string) *SessionRecorder { + return NewSessionRecorder(RecorderOptions{ + Store: r.store, + SessionID: r.stream.SessionID, + RunID: runID, + Actor: actor, + Parent: &CausalParent{ + RunID: r.runID, + CausedByEventID: causedByEventID, + }, + }) +} + +func (r *SessionRecorder) appendEvent(ctx context.Context, event ProposedEvent) (StoredEvent, error) { + r.mu.Lock() + defer r.mu.Unlock() receipt, err := r.store.Append(ctx, r.stream, r.expectedVersion, NewID(), event) if err != nil { return StoredEvent{}, err @@ -127,3 +162,10 @@ func errorPayload(err error) map[string]any { } return map[string]any{"error": err.Error()} } + +func payloadOrEmpty(payload map[string]any) map[string]any { + if payload == nil { + return map[string]any{} + } + return payload +} diff --git a/go/store_test.go b/go/store_test.go index 12f0047..079a1ad 100644 --- a/go/store_test.go +++ b/go/store_test.go @@ -96,3 +96,44 @@ func TestResumeRecorderRejectsExpectedVersion(t *testing.T) { t.Fatal("resume accepted an explicit expected version") } } + +func TestOrchestratorLinksMultipleAgentRuns(t *testing.T) { + ctx := context.Background() + store := NewMemoryEventStore() + orchestrator := NewSessionRecorder(RecorderOptions{ + Store: store, SessionID: "session", RunID: "orchestrator-run", + Actor: Actor{Type: "orchestrator", ID: "planner"}, + }) + if _, err := orchestrator.StartRun(ctx, nil); err != nil { + t.Fatalf("start orchestrator: %v", err) + } + for _, role := range []string{"researcher", "reviewer"} { + dispatch, err := orchestrator.Record( + ctx, "orchestration.agent.dispatched", map[string]any{"role": role}, "", "", + ) + if err != nil { + t.Fatalf("record %s dispatch: %v", role, err) + } + child := orchestrator.Child(role+"-run", Actor{Type: "agent", ID: role}, dispatch.EventID) + if _, err := child.StartRun(ctx, nil); err != nil { + t.Fatalf("start %s: %v", role, err) + } + } + + var childRuns []string + for event, err := range store.ScanSession(ctx, "session", "") { + if err != nil { + t.Fatalf("scan session: %v", err) + } + if event.ParentRunID == "" { + continue + } + if event.ParentRunID != "orchestrator-run" || event.CausedByEventID == "" { + t.Fatalf("invalid causal edge: %#v", event.ProposedEvent) + } + childRuns = append(childRuns, event.RunID) + } + if len(childRuns) != 2 || childRuns[0] != "researcher-run" || childRuns[1] != "reviewer-run" { + t.Fatalf("child runs = %v", childRuns) + } +} diff --git a/go/types.go b/go/types.go index 8547184..5e28d2d 100644 --- a/go/types.go +++ b/go/types.go @@ -13,6 +13,11 @@ type EventStream struct { StreamID string `json:"stream_id"` } +type CausalParent struct { + RunID string + CausedByEventID string +} + type ProposedEvent struct { SchemaVersion string `json:"schema_version"` EventID string `json:"event_id"` diff --git a/python/examples/orchestrated_agents.py b/python/examples/orchestrated_agents.py new file mode 100644 index 0000000..46bcdaf --- /dev/null +++ b/python/examples/orchestrated_agents.py @@ -0,0 +1,47 @@ +import asyncio +from uuid import uuid4 + +from agent_ledger import Actor, SessionRecorder, inspect_session +from agent_ledger.stores.memory import MemoryEventStore + + +async def main() -> None: + store = MemoryEventStore() + session_id = str(uuid4()) + orchestrator = SessionRecorder( + store=store, + session_id=session_id, + run_id="orchestrator-1", + actor=Actor(type="orchestrator", id="planner"), + ) + await orchestrator.start_run(payload={"goal": "research and review a proposal"}) + + for role in ("researcher", "reviewer"): + dispatch = await orchestrator.record( + "orchestration.agent.dispatched", + payload={"role": role}, + ) + agent = orchestrator.child( + run_id=f"{role}-1", + actor=Actor(type="agent", id=role, framework="plain-loop"), + caused_by_event_id=dispatch.event_id, + ) + await agent.start_run( + payload={ + "agent": {"id": role, "version": "1"}, + "framework": {"name": "plain-loop"}, + "code": {"revision": "example"}, + } + ) + step_id = str(uuid4()) + await agent.start_step(step_id, payload={"role": role}) + await agent.complete_step(step_id) + await agent.complete_run() + + events = [event async for event in store.scan_session(session_id)] + inspection = inspect_session(events) + print(f"events={len(inspection.timeline)} run_edges={len(inspection.run_edges)}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/tests/test_models.py b/python/tests/test_models.py index d14d23d..17405f8 100644 --- a/python/tests/test_models.py +++ b/python/tests/test_models.py @@ -3,7 +3,9 @@ import json from pathlib import Path +import pytest from jsonschema import Draft202012Validator, FormatChecker +from jsonschema.exceptions import ValidationError from agent_ledger import Actor, MemoryArtifactStore, ProposedEvent, StoredEvent from agent_ledger.frameworks.plain_loop import PlainLoopProfile @@ -32,6 +34,25 @@ def test_event_matches_normative_json_schema() -> None: validator.validate(stored.model_dump(mode="json", exclude_none=True)) +def test_event_schema_requires_complete_causal_parent() -> None: + schema_path = Path(__file__).parents[2] / "spec" / "schemas" / "event.schema.json" + schema = json.loads(schema_path.read_text()) + event = { + "schema_version": "1.0", + "event_id": "event", + "event_type": "run.started", + "session_id": "session", + "run_id": "run", + "actor": {"type": "agent", "id": "child"}, + "occurred_at": "2026-08-09T00:00:00Z", + "parent_run_id": "parent", + "payload": {}, + } + + with pytest.raises(ValidationError): + Draft202012Validator(schema).validate(event) + + async def test_memory_artifact_round_trip() -> None: store = MemoryArtifactStore() ref = await store.put("session", b"large model output", "text/plain") diff --git a/python/tests/test_recorder.py b/python/tests/test_recorder.py index 74ffae8..e179bff 100644 --- a/python/tests/test_recorder.py +++ b/python/tests/test_recorder.py @@ -54,24 +54,37 @@ async def test_retry_keeps_step_and_gets_new_attempt() -> None: assert inspection.unresolved_attempts[0].step_id == "step-1" -async def test_child_run_forms_causal_edge() -> None: +async def test_orchestrator_links_multiple_agent_runs() -> None: store = MemoryEventStore() - parent = _recorder(store) - trigger = await parent.start_step("delegate") - child = parent.child( - run_id=str(uuid4()), - actor=Actor(type="agent", id="child"), - caused_by_event_id=trigger.event_id, + parent = SessionRecorder( + store=store, + session_id=str(uuid4()), + run_id="orchestrator-run", + actor=Actor(type="orchestrator", id="planner"), ) - await child.start_run() + await parent.start_run() + children: list[SessionRecorder] = [] + for role in ("researcher", "reviewer"): + trigger = await parent.record( + "orchestration.agent.dispatched", + payload={"role": role}, + ) + child = parent.child( + run_id=f"{role}-run", + actor=Actor(type="agent", id=role), + caused_by_event_id=trigger.event_id, + ) + await child.start_run() + children.append(child) events = [event async for event in store.scan_session(parent.stream.session_id)] inspection = inspect_session(events) - assert len(inspection.run_edges) == 1 - assert inspection.run_edges[0].parent_run_id == parent.run_id - assert inspection.run_edges[0].child_run_id == child.run_id - assert inspection.run_edges[0].caused_by_event_id == trigger.event_id + assert len(inspection.run_edges) == 2 + assert {edge.parent_run_id for edge in inspection.run_edges} == {parent.run_id} + assert {edge.child_run_id for edge in inspection.run_edges} == { + child.run_id for child in children + } async def test_plain_loop_profile_restores_snapshot_and_tail() -> None: diff --git a/spec/rfcs/0001-agent-ledger.md b/spec/rfcs/0001-agent-ledger.md index bd122ee..d36b616 100644 --- a/spec/rfcs/0001-agent-ledger.md +++ b/spec/rfcs/0001-agent-ledger.md @@ -8,24 +8,41 @@ Draft specification for the `0.x` library line. Agent loops can be interrupted by process replacement, autoscaling, model errors, and rate limits. Framework checkpoints solve recovery inside one runtime, but they do not necessarily provide a -framework-neutral timeline across distributed agents. Agent Ledger defines the durable facts that -framework integrations can share while leaving framework state restoration to those integrations. +framework-neutral timeline across distributed agents and their orchestrator. Agent Ledger defines +the durable execution facts that all of those producers can share while leaving framework state +restoration and workflow control to their respective owners. ## Concepts | Concept | Meaning | | --- | --- | -| Session | End-to-end task and the boundary of the global timeline. | +| Session | End-to-end task and the boundary of the global timeline. It is not a framework-native chat session. | | Event Stream | One optimistic-concurrency partition inside a session. | +| Run | One semantic execution by an agent, orchestrator, or other actor. | | Step | Logical work that survives retries. | | Attempt | One physical model or tool call within a step. | | Event | Immutable fact proposed by a producer and enriched by a store. | | Framework Adapter | Recording and recovery bindings for one framework. | | Trajectory | Read-side projection for evaluation or analysis. | -A child agent starts another run in the same session. Its `run.started` event carries -`parent_run_id` and `caused_by_event_id`. The causal DAG, rather than wall-clock order, is the -authoritative relationship between runs. +When an orchestrator or agent delegates work, the child actor starts another run in the same +session. Its `run.started` event carries `parent_run_id` and `caused_by_event_id`. The causal DAG, +rather than wall-clock order, is the authoritative relationship between runs. + +## Layer boundary + +Agent Ledger crosses the Agent Loop and Orchestrator layers without owning either one: + +- an agent loop records steps, model and tool attempts, outcomes, and framework-native recovery + state; +- an orchestrator records decisions, delegation, approvals, and execution outcomes while retaining + ownership of desired state, scheduling, leases, and reconciliation; +- stores own atomic persistence, not agent or workflow semantics; +- read-side consumers derive recovery input, timelines, trajectories, alerts, and evaluation data. + +The ledger does not promote events into memory, change prompts or skills, decide whether a result +is correct, or activate a new capability version. Those systems may record their decisions in the +same session, but their policies remain outside the core library. ## Event envelope @@ -39,6 +56,22 @@ establishes causality. Large inputs and outputs should use an `ArtifactRef`. The event keeps the content digest, media type, byte size, and URI while an application-selected `ArtifactStore` owns the bytes. +Core normalized event families use `session.*`, `run.*`, `step.*`, `model.*`, and `tool.*`. +Framework-native records use `framework..*`. Orchestrators and applications may add +namespaced event types such as `orchestration.agent.dispatched`; readers preserve unknown event +types and payloads. + +### Run provenance + +Analysis needs to know which system configuration produced a run. A `run.started` payload should +therefore identify the agent or orchestrator implementation, framework and adapter versions, model, +code revision, and relevant prompt, skill, toolset, dataset, or verifier versions or digests when +they are known. + +Provenance stays in the run-start payload instead of being repeated in every envelope. Consumers +join it to later events by `run_id`. V1 deliberately leaves the payload open while real adapters +establish which references are portable enough to standardize. + ## Append contract The empty stream version is `-1`; the first stored event has version `0`. A stream is identified by @@ -98,6 +131,19 @@ The bundled plain-loop profile demonstrates snapshot recovery. Frameworks with n checkpointing preserve that state losslessly and restore with their own APIs. RFC 0002 defines the recording/recovery split and capability declarations. +## Read models and projections + +The append log is the source of truth; recovery and analysis are independent projections: + +- a framework adapter combines native records with normalized attempts to restore its own context; +- a session timeline merges all event streams for display, then uses causal links to explain the + relationship between actors; +- trajectory exporters select normalized steps and attempts for evaluation or training; +- memory and capability-improvement pipelines consume only facts accepted by their own validation + and approval policies. + +No projection may rewrite historical events or claim that commit order establishes causality. + ## Store durability An append receipt means the selected store accepted the transaction. End-to-end durability still diff --git a/spec/rfcs/0002-polyglot-adapters.md b/spec/rfcs/0002-polyglot-adapters.md index e577717..c995a47 100644 --- a/spec/rfcs/0002-polyglot-adapters.md +++ b/spec/rfcs/0002-polyglot-adapters.md @@ -65,6 +65,10 @@ The strict Pi integration implements its native `SessionStorage`. Entries and ac are stored losslessly, then Pi's `Session` rebuilds model context. Direct AgentHarness hooks add normalized model, tool, turn, and run events. +Pi's append-only entry tree is useful native recovery state, but it is not the portable Agent +Ledger session model. Other frameworks and orchestrators are not required to expose Pi entry types, +leaf selection, or branching semantics. + A coding-agent extension that swallows model-hook errors must declare model prewrite as `best_effort`. It can still emit telemetry, but it is not the strict profile. diff --git a/spec/schemas/event.schema.json b/spec/schemas/event.schema.json index f9fdce3..3a592dd 100644 --- a/spec/schemas/event.schema.json +++ b/spec/schemas/event.schema.json @@ -32,6 +32,10 @@ "commit_cursor": { "type": "string", "minLength": 1 }, "committed_at": { "type": "string", "format": "date-time" } }, + "dependentRequired": { + "parent_run_id": ["caused_by_event_id"], + "caused_by_event_id": ["parent_run_id"] + }, "additionalProperties": false, "$defs": { "actor": { diff --git a/typescript/packages/core/src/recorder.ts b/typescript/packages/core/src/recorder.ts index cf4eb1c..d0f9510 100644 --- a/typescript/packages/core/src/recorder.ts +++ b/typescript/packages/core/src/recorder.ts @@ -1,12 +1,18 @@ import type { EventStore } from "./store.js"; import { proposedEvent, type Actor, type AttemptHandle, type EventStream, type JsonValue, type StoredEvent } from "./types.js"; +export interface CausalParent { + runId: string; + causedByEventId: string; +} + export interface RecorderOptions { store: EventStore; sessionId: string; runId: string; actor: Actor; streamId?: string; + parent?: CausalParent; expectedVersion?: number; } @@ -15,6 +21,7 @@ export class SessionRecorder { readonly stream: EventStream; readonly runId: string; readonly actor: Actor; + readonly parent: CausalParent | undefined; #expectedVersion: number; #tail = Promise.resolve(); @@ -22,6 +29,7 @@ export class SessionRecorder { this.store = options.store; this.runId = options.runId; this.actor = options.actor; + this.parent = options.parent === undefined ? undefined : { ...options.parent }; this.stream = { session_id: options.sessionId, stream_id: options.streamId ?? options.runId }; this.#expectedVersion = options.expectedVersion ?? -1; } @@ -39,9 +47,14 @@ export class SessionRecorder { payload?: { [key: string]: JsonValue }; stepId?: string; attemptId?: string; + parentRunId?: string; + causedByEventId?: string; appendId?: string; } = {}, ): Promise { + if ((options.parentRunId === undefined) !== (options.causedByEventId === undefined)) { + throw new Error("parentRunId and causedByEventId must be set together"); + } return this.#serialize(async () => { const event = proposedEvent({ event_type: eventType, @@ -50,6 +63,8 @@ export class SessionRecorder { actor: this.actor, ...(options.stepId === undefined ? {} : { step_id: options.stepId }), ...(options.attemptId === undefined ? {} : { attempt_id: options.attemptId }), + ...(options.parentRunId === undefined ? {} : { parent_run_id: options.parentRunId }), + ...(options.causedByEventId === undefined ? {} : { caused_by_event_id: options.causedByEventId }), payload: options.payload ?? {}, }); const receipt = await this.store.append( @@ -69,6 +84,31 @@ export class SessionRecorder { }); } + startRun(payload: { [key: string]: JsonValue } = {}): Promise { + return this.record("run.started", { + payload, + ...(this.parent === undefined + ? {} + : { parentRunId: this.parent.runId, causedByEventId: this.parent.causedByEventId }), + }); + } + + child(options: { + runId: string; + actor: Actor; + causedByEventId: string; + streamId?: string; + }): SessionRecorder { + return new SessionRecorder({ + store: this.store, + sessionId: this.stream.session_id, + runId: options.runId, + actor: options.actor, + ...(options.streamId === undefined ? {} : { streamId: options.streamId }), + parent: { runId: this.runId, causedByEventId: options.causedByEventId }, + }); + } + async beforeModelCall(stepId: string, payload: { [key: string]: JsonValue }): Promise { return this.#beforeCall("model", stepId, payload); } diff --git a/typescript/packages/core/test/core.test.ts b/typescript/packages/core/test/core.test.ts index b005efe..5cb5c71 100644 --- a/typescript/packages/core/test/core.test.ts +++ b/typescript/packages/core/test/core.test.ts @@ -3,7 +3,13 @@ import { readFile } from "node:fs/promises"; import { resolve } from "node:path"; import { test } from "node:test"; -import { canonicalAppendDigest, DuplicateEvent, MemoryEventStore, proposedEvent } from "../src/index.js"; +import { + canonicalAppendDigest, + DuplicateEvent, + MemoryEventStore, + proposedEvent, + SessionRecorder, +} from "../src/index.js"; test("event streams order framework state independently from semantic runs", async () => { const store = new MemoryEventStore(); @@ -70,3 +76,34 @@ test("duplicate event ids reject the whole append batch", async () => { for await (const event of store.readStream(stream)) stored.push(event); assert.deepEqual(stored, []); }); + +test("an orchestrator links multiple agent runs in one session", async () => { + const store = new MemoryEventStore(); + const orchestrator = new SessionRecorder({ + store, + sessionId: "session", + runId: "orchestrator-run", + actor: { type: "orchestrator", id: "planner" }, + }); + await orchestrator.startRun(); + + for (const role of ["researcher", "reviewer"]) { + const dispatch = await orchestrator.record("orchestration.agent.dispatched", { + payload: { role }, + }); + const child = orchestrator.child({ + runId: `${role}-run`, + actor: { type: "agent", id: role }, + causedByEventId: dispatch.event_id, + }); + await child.startRun(); + } + + const childStarts = []; + for await (const event of store.scanSession("session")) { + if (event.parent_run_id !== undefined) childStarts.push(event); + } + assert.deepEqual(childStarts.map((event) => event.run_id), ["researcher-run", "reviewer-run"]); + assert.deepEqual(childStarts.map((event) => event.parent_run_id), ["orchestrator-run", "orchestrator-run"]); + assert.ok(childStarts.every((event) => event.caused_by_event_id !== undefined)); +}); diff --git a/typescript/packages/pi/src/harness.ts b/typescript/packages/pi/src/harness.ts index 1f0e1d9..15daab7 100644 --- a/typescript/packages/pi/src/harness.ts +++ b/typescript/packages/pi/src/harness.ts @@ -67,7 +67,10 @@ export function bindPiHarness(harness: PiHarnessLike, recorder: SessionRecorder) disposers.push(harness.subscribe(async (event) => { switch (event.type) { case "agent_start": - await recorder.record("run.started", { payload: { adapter: PI_ADAPTER.adapter_id } }); + await recorder.startRun({ + adapter: { id: PI_ADAPTER.adapter_id, version: PI_ADAPTER.adapter_version }, + framework: { name: PI_ADAPTER.framework }, + }); break; case "turn_start": turn += 1;