feat: support orchestrated agent sessions

Author: liqiankun1111

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.

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 = &copy
+	}
 	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
+}

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)
+	}
+}

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"`

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())

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")

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: