From e5c42de06d163c0d3f3bd3b1a0fdafee1ddb9f3e Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 14 Nov 2026 01:20:06 +0000 Subject: [PATCH] Include conversation ancestry in execution observations --- docs/roadmap/implementation.md | 2 + engine.go | 3 ++ internal/observation/recorder.go | 6 ++- internal/usecase/prepared_execution.go | 1 + observation_contract_test.go | 67 ++++++++++++++++++++++++++ observations.go | 2 +- 6 files changed, 79 insertions(+), 2 deletions(-) diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 7c91846..f66f5e7 100644 --- a/docs/roadmap/implementation.md +++ b/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 diff --git a/engine.go b/engine.go index fe436b6..8a55ccc 100644 --- a/engine.go +++ b/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) diff --git a/internal/observation/recorder.go b/internal/observation/recorder.go index 29a4ffb..fb24b70 100644 --- a/internal/observation/recorder.go +++ b/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, } diff --git a/internal/usecase/prepared_execution.go b/internal/usecase/prepared_execution.go index 7ac526c..22fa2f5 100644 --- a/internal/usecase/prepared_execution.go +++ b/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 diff --git a/observation_contract_test.go b/observation_contract_test.go index 4e38214..0d6d09a 100644 --- a/observation_contract_test.go +++ b/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 diff --git a/observations.go b/observations.go index bab330d..3e28f8a 100644 --- a/observations.go +++ b/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,