Include conversation ancestry in execution observations

docs/roadmap/implementation.md

@@ -392,6 +392,8 @@ Do not add a separate preparation or validation algorithm for continuation.
 
 ## Stage 4 — Propagate Ancestry Through Ordinary Observations
 
+**Status:** Complete.
+
 **Scope:** immediate-parent metadata without accounting changes.
 
 1. Add parent metadata to the internal recorder's event construction and the root

engine.go

@@ -514,6 +514,9 @@ func (e *Engine) Run(ctx context.Context, req RunRequest) (*RunResult, error) {
 		return e.run(ctx, req)
 	}
 	recorder, idErr := observation.New(func(event domain.ExecutionEvent) { req.Observer(fromDomainExecutionEvent(event)) })
+	if req.ContinueFrom != nil && req.ContinueFrom.internal != nil {
+		recorder.SetParentRunID(req.ContinueFrom.internal.RunID())
+	}
 	recorder.Start()
 	if idErr != nil {
 		err := fmt.Errorf("failed to create run id: %w", idErr)

internal/observation/recorder.go

@@ -16,6 +16,7 @@ import (
 // calls methods in execution order; it has no shared history or background work.
 type Recorder struct {
 	runID                string
+	parentRunID          string
 	startedAt            time.Time
 	now                  func() time.Time
 	observer             domain.ExecutionObserver
@@ -95,6 +96,9 @@ func NewRunID() (string, error) {
 // RunID returns this invocation's identity.
 func (r *Recorder) RunID() string { return r.runID }
 
+// SetParentRunID records the immediate source identity before start delivery.
+func (r *Recorder) SetParentRunID(id string) { r.parentRunID = id }
+
 // Start delivers the opening event with only identities already resolved.
 func (r *Recorder) Start() {
 	event := r.event(domain.ExecutionStarted)
@@ -177,7 +181,7 @@ func (r *Recorder) Finish(data FinishData) {
 
 func (r *Recorder) event(kind domain.ExecutionEventKind) domain.ExecutionEvent {
 	return domain.ExecutionEvent{
-		Kind: kind, RunID: r.runID, Time: r.now(), PromptID: r.selection.PromptID,
+		Kind: kind, RunID: r.runID, ParentRunID: r.parentRunID, Time: r.now(), PromptID: r.selection.PromptID,
 		ProfileID: r.selection.ProfileID,
 		BackendID: r.selection.BackendID, Model: r.selection.Model,
 	}

internal/usecase/prepared_execution.go

@@ -112,6 +112,7 @@ func (r *Runner) RunPrepared(ctx context.Context, prepared *PreparedExecution) (
 	if payload.observer != nil {
 		var idErr error
 		recorder, idErr = observation.New(payload.observer)
+		recorder.SetParentRunID(payload.parent.RunID())
 		observation.RegisterClaim(ctx, recorder)
 		ctx = observation.WithRecorder(ctx, recorder)
 		selected := payload.prepared

observation_contract_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"strings"
 	"testing"
 
@@ -15,6 +16,72 @@ func observationEngine(t *testing.T, client promptkit.LLMClient) *promptkit.Engi
 	return newSimpleEngine(t, "hello", promptkit.WithLLMClient(client))
 }
 
+func TestContinuationObservationAncestry(t *testing.T) {
+	client := &fakeLLMClient{response: &promptkit.GenerateResponse{Content: "private answer"}}
+	engine := observationEngine(t, client)
+	ctx := context.Background()
+	initial, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var events []promptkit.ExecutionEvent
+	observer := func(event promptkit.ExecutionEvent) { events = append(events, event) }
+	req := promptkit.RunRequest{PromptID: "prompt", ContinueFrom: initial.Exchange, Observer: observer}
+	prepared, err := engine.PrepareExecution(ctx, req)
+	if err != nil || len(events) != 0 {
+		t.Fatalf("preparation events=%#v error=%v", events, err)
+	}
+	check := func(result *promptkit.RunResult, runErr error) {
+		t.Helper()
+		if runErr != nil || result == nil || len(events) != 3 {
+			t.Fatalf("result=%#v error=%v events=%#v", result, runErr, events)
+		}
+		for _, event := range events {
+			if event.ParentRunID != initial.RunID || event.RunID != result.RunID || strings.Contains(fmt.Sprintf("%+v", event), "private answer") {
+				t.Fatalf("event = %#v", event)
+			}
+		}
+		events = nil
+	}
+	result, err := engine.Run(ctx, req)
+	check(result, err)
+	result, err = engine.RunPrepared(ctx, prepared)
+	check(result, err)
+	client.err = errors.New("failure")
+	if result, err := engine.Run(ctx, req); result != nil || err == nil || len(events) != 3 {
+		t.Fatalf("failure result=%#v error=%v events=%#v", result, err, events)
+	}
+	for _, event := range events {
+		if event.ParentRunID != initial.RunID {
+			t.Fatalf("failure event = %#v", event)
+		}
+	}
+}
+
+func TestContinuationNestedRunHasIndependentAncestry(t *testing.T) {
+	engine := observationEngine(t, &fakeLLMClient{response: &promptkit.GenerateResponse{Content: "ok"}})
+	ctx := context.Background()
+	initial, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", CaptureExchange: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	var nested []promptkit.ExecutionEvent
+	var nestedErr error
+	_, err = engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", ContinueFrom: initial.Exchange, Observer: func(event promptkit.ExecutionEvent) {
+		if event.Kind == promptkit.ExecutionFinished {
+			_, nestedErr = engine.Run(ctx, promptkit.RunRequest{PromptID: "prompt", Observer: func(value promptkit.ExecutionEvent) { nested = append(nested, value) }})
+		}
+	}})
+	if err != nil || nestedErr != nil || len(nested) == 0 {
+		t.Fatalf("errors=%v, %v; events=%#v", err, nestedErr, nested)
+	}
+	for _, event := range nested {
+		if event.ParentRunID != "" {
+			t.Fatalf("nested event = %#v", event)
+		}
+	}
+}
+
 func TestRunObservationCoversFacadeErrors(t *testing.T) {
 	for _, tc := range []struct {
 		name    string

observations.go

@@ -130,7 +130,7 @@ func fromDomainExecutionEvent(value domain.ExecutionEvent) ExecutionEvent {
 		return ObservedTokenCount{Value: c.Value, Known: c.Known, Complete: c.Complete}
 	}
 	return ExecutionEvent{
-		Kind: ExecutionEventKind(value.Kind), RunID: value.RunID, Time: value.Time,
+		Kind: ExecutionEventKind(value.Kind), RunID: value.RunID, ParentRunID: value.ParentRunID, Time: value.Time,
 		PromptID: value.PromptID, ProfileID: value.ProfileID,
 		BackendID: value.BackendID, Model: value.Model, Generation: value.Generation,
 		Attempt: value.Attempt, Dispatched: value.Dispatched, CapacityWait: value.CapacityWait,