diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 843d133..bf7ca2c 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -465,6 +465,11 @@ attempt, retry-success, exhausted retries, cancellation during backoff, and canceled reacquisition. Check evidence retention and timing/accounting; existing retry tests own unchanged retry policy. Run observation tests after extraction. +**Completed: 2026-10-13.** Added built-in generation attempt evidence and shared +presence-aware usage accumulation. Validation passed: `go test -race +./internal/generation ./internal/observation`, `go vet ./...`, `go test ./...`, +and `git diff --check`. + ## Stage 10: Retain Injected-Client Evidence Without HTTP Claims Scope: the injected branch of `internal/generation`. diff --git a/internal/generation/coordinator.go b/internal/generation/coordinator.go index 89d9e89..ebb56d4 100644 --- a/internal/generation/coordinator.go +++ b/internal/generation/coordinator.go @@ -72,14 +72,41 @@ func (e *ExhaustedError) Attempts() uint64 { // Generate executes either one opaque client call or a frozen built-in HTTP // sequence with one permit per actual dispatch. func (c *Coordinator) Generate(ctx context.Context, req domain.GenerateRequest) (*domain.GenerateResponse, error) { + response, err, _ := c.generate(ctx, req, false) + return response, err +} + +// GenerateWithEvidence runs the built-in retry path and returns its factual +// attempt evidence alongside any response retained before cancellation. +func (c *Coordinator) GenerateWithEvidence( + ctx context.Context, + req domain.GenerateRequest, +) (*domain.GenerateResponse, error, GenerationEvidence) { + return c.generate(ctx, req, true) +} + +func (c *Coordinator) generate( + ctx context.Context, + req domain.GenerateRequest, + captureEvidence bool, +) (*domain.GenerateResponse, error, GenerationEvidence) { if c.injected != nil { - return c.generateInjected(ctx, req) + response, err := c.generateInjected(ctx, req) + return response, err, GenerationEvidence{} } prepared, err := c.builtIn.PrepareGeneration(req) if err != nil { - return nil, err + return nil, err, GenerationEvidence{} } + evidence := GenerationEvidence{} + var usage observation.UsageAccumulator + finish := func(response *domain.GenerateResponse, err error) (*domain.GenerateResponse, error, GenerationEvidence) { + if captureEvidence { + evidence.Usage = usage.Observed() + } + return response, err, evidence + } var attempts, retriesUsed uint64 recorder := observation.FromContext(ctx) generation := req.Generation @@ -88,50 +115,82 @@ func (c *Coordinator) Generate(ctx context.Context, req domain.GenerateRequest) } for { if err := ctx.Err(); err != nil { - return nil, err + return finish(nil, err) } ordinal := attempts + 1 - outcome, dispatched, capacityWait, generationDuration := c.execute(ctx, req.Target.BackendID, prepared) + outcome, dispatched, capacityWait, generationDuration := c.execute(ctx, req.Target.BackendID, prepared, captureEvidence) if dispatched { attempts++ } + if captureEvidence { + attempt := AttemptEvidence{ + Dispatched: dispatched, + HTTP: cloneHTTPEvidence(outcome.Evidence), + CapacityWait: capacityWait, + RequestDuration: generationDuration, + Failure: ClassifyFailure(outcome.Err), + ResponseDecoded: outcome.ResponseDecoded, + GenerationCompleted: outcome.GenerationCompleted, + } + if outcome.StructuredOutputRequested != nil { + requested := *outcome.StructuredOutputRequested + attempt.StructuredOutputRequested = &requested + } + if dispatched { + usage.Add(outcome.Usage, outcome.UsagePresence) + var attemptUsage observation.UsageAccumulator + attemptUsage.Add(outcome.Usage, outcome.UsagePresence) + attempt.Usage = attemptUsage.Observed() + } + evidence.Attempts = append(evidence.Attempts, attempt) + } if recorder != nil { recorder.Attempt(observation.AttemptData{Generation: generation, Attempt: ordinal, Dispatched: dispatched, CapacityWait: capacityWait, GenerationDuration: generationDuration, - Outcome: attemptOutcome(outcome.Err), Failure: attemptFailure(outcome.Err), + Outcome: attemptOutcome(outcome.Err), Failure: ClassifyFailure(outcome.Err), StatusCode: outcome.StatusCode, Usage: outcome.Usage, UsagePresence: outcome.UsagePresence}) } if outcome.Err == nil { if err := ctx.Err(); err != nil { - return nil, err + if captureEvidence { + return finish(outcome.Response, err) + } + return finish(nil, err) } - return outcome.Response, nil + return finish(outcome.Response, nil) } if !dispatched || !eligibleStatus(outcome) { - return nil, outcome.Err + if captureEvidence { + return finish(outcome.Response, outcome.Err) + } + return finish(nil, outcome.Err) } if retriesUsed == req.TransientRetries { if retriesUsed == 0 { - return nil, outcome.Err + return finish(nil, outcome.Err) } - return nil, &ExhaustedError{attempts: attempts, cause: outcome.Err} + return finish(nil, &ExhaustedError{attempts: attempts, cause: outcome.Err}) } if err := ctx.Err(); err != nil { - return nil, err + return finish(nil, err) } retryOrdinal := retriesUsed + 1 delay := retryDelay(c.backoff(retryOrdinal), outcome.RetryNotBefore, c.now()) waitStart := c.now() waitErr := c.wait(ctx, delay) + waitDuration := elapsed(waitStart, c.now()) + if captureEvidence { + evidence.RetryDelays = append(evidence.RetryDelays, waitDuration) + } if recorder != nil { recorder.Delay(observation.DelayData{Generation: generation, Attempt: attempts, - Elapsed: elapsed(waitStart, c.now()), Failure: attemptFailure(waitErr)}) + Elapsed: waitDuration, Failure: ClassifyFailure(waitErr)}) } if waitErr != nil { - return nil, waitErr + return finish(nil, waitErr) } if err := ctx.Err(); err != nil { - return nil, err + return finish(nil, err) } retriesUsed++ } @@ -148,7 +207,7 @@ func (c *Coordinator) generateInjected(ctx context.Context, req domain.GenerateR capacityWait := c.capacityWait(req.Target.BackendID, start) if err != nil { if recorder != nil { - recorder.Attempt(observation.AttemptData{Generation: generation, Attempt: 1, CapacityWait: capacityWait, Outcome: domain.ExecutionOutcomeError, Failure: attemptFailure(err)}) + recorder.Attempt(observation.AttemptData{Generation: generation, Attempt: 1, CapacityWait: capacityWait, Outcome: domain.ExecutionOutcomeError, Failure: ClassifyFailure(err)}) } return nil, err } @@ -156,7 +215,7 @@ func (c *Coordinator) generateInjected(ctx context.Context, req domain.GenerateR if err := ctx.Err(); err != nil { release() if recorder != nil { - recorder.Attempt(observation.AttemptData{Generation: generation, Attempt: 1, CapacityWait: capacityWait, Outcome: domain.ExecutionOutcomeError, Failure: attemptFailure(err)}) + recorder.Attempt(observation.AttemptData{Generation: generation, Attempt: 1, CapacityWait: capacityWait, Outcome: domain.ExecutionOutcomeError, Failure: ClassifyFailure(err)}) } return nil, err } @@ -169,7 +228,7 @@ func (c *Coordinator) generateInjected(ctx context.Context, req domain.GenerateR release() if recorder != nil { data := observation.AttemptData{Generation: generation, Attempt: 1, Dispatched: true, CapacityWait: capacityWait, GenerationDuration: duration, - Outcome: attemptOutcome(err), Failure: attemptFailure(err)} + Outcome: attemptOutcome(err), Failure: ClassifyFailure(err)} if err == nil { data.Usage = response.Usage data.UsagePresence = response.UsagePresence @@ -185,7 +244,7 @@ func (c *Coordinator) generateInjected(ctx context.Context, req domain.GenerateR return response, nil } -func (c *Coordinator) execute(ctx context.Context, backendID string, prepared *llm.PreparedGeneration) (outcome llm.AttemptOutcome, dispatched bool, capacityWait, generationDuration time.Duration) { +func (c *Coordinator) execute(ctx context.Context, backendID string, prepared *llm.PreparedGeneration, captureEvidence bool) (outcome llm.AttemptOutcome, dispatched bool, capacityWait, generationDuration time.Duration) { start := c.now() release, err := c.manager.Acquire(ctx, backendID) capacityWait = c.capacityWait(backendID, start) @@ -197,8 +256,14 @@ func (c *Coordinator) execute(ctx context.Context, backendID string, prepared *l return llm.AttemptOutcome{Err: err}, false, capacityWait, 0 } dispatchStart := c.now() - outcome = prepared.Execute(ctx) - return outcome, true, capacityWait, elapsed(dispatchStart, c.now()) + if captureEvidence { + outcome = prepared.ExecuteWithEvidence(ctx) + dispatched = outcome.Evidence.DispatchInvoked + } else { + outcome = prepared.Execute(ctx) + dispatched = true + } + return outcome, dispatched, capacityWait, elapsed(dispatchStart, c.now()) } func (c *Coordinator) capacityWait(backendID string, start time.Time) time.Duration { @@ -222,7 +287,8 @@ func attemptOutcome(err error) domain.ExecutionOutcome { return domain.ExecutionOutcomeError } -func attemptFailure(err error) domain.ExecutionFailureKind { +// ClassifyFailure maps an execution error to the shared content-free category. +func ClassifyFailure(err error) domain.ExecutionFailureKind { if err == nil { return domain.ExecutionFailureNone } diff --git a/internal/generation/evidence.go b/internal/generation/evidence.go new file mode 100644 index 0000000..7685e6d --- /dev/null +++ b/internal/generation/evidence.go @@ -0,0 +1,38 @@ +package generation + +import ( + "time" + + "gitea.maximumdirect.net/eric/promptkit/internal/domain" + "gitea.maximumdirect.net/eric/promptkit/internal/llm" +) + +// GenerationEvidence contains invocation-local request evidence without raw +// provider bodies, headers, credentials, or generated content. +type GenerationEvidence struct { + Attempts []AttemptEvidence + RetryDelays []time.Duration + Usage domain.ObservedTokenUsage +} + +// AttemptEvidence describes one capacity acquisition and any dispatch it +// reached. Generated content remains on the separate response return value. +type AttemptEvidence struct { + Dispatched bool + HTTP llm.HTTPExecutionEvidence + CapacityWait time.Duration + RequestDuration time.Duration + Failure domain.ExecutionFailureKind + Usage domain.ObservedTokenUsage + ResponseDecoded bool + GenerationCompleted bool + StructuredOutputRequested *bool +} + +func cloneHTTPEvidence(evidence llm.HTTPExecutionEvidence) llm.HTTPExecutionEvidence { + if evidence.HTTPStatus != nil { + status := *evidence.HTTPStatus + evidence.HTTPStatus = &status + } + return evidence +} diff --git a/internal/generation/evidence_test.go b/internal/generation/evidence_test.go new file mode 100644 index 0000000..a658ba0 --- /dev/null +++ b/internal/generation/evidence_test.go @@ -0,0 +1,188 @@ +package generation + +import ( + "context" + "errors" + "net/http" + "sync" + "sync/atomic" + "testing" + "time" + + "gitea.maximumdirect.net/eric/promptkit/internal/domain" +) + +func TestGenerateWithEvidenceRetainsAttemptsAndUsage(t *testing.T) { + var now = time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) + calls := 0 + coordinator := testCoordinator(t, func(*http.Request) (*http.Response, error) { + calls++ + if calls == 1 { + now = now.Add(4 * time.Millisecond) + return response(http.StatusServiceUnavailable, `{"error":{}}`), nil + } + now = now.Add(7 * time.Millisecond) + return response(http.StatusOK, `{"choices":[{"message":{"content":"ok"}}],"usage":{"prompt_tokens":3}}`), nil + }) + coordinator.now = func() time.Time { return now } + coordinator.wait = func(context.Context, time.Duration) error { + now = now.Add(19 * time.Millisecond) + return nil + } + + result, err, evidence := coordinator.GenerateWithEvidence(context.Background(), testRequest(1)) + if err != nil || result == nil || result.Content != "ok" || calls != 2 { + t.Fatalf("result=%+v error=%v calls=%d", result, err, calls) + } + if len(evidence.Attempts) != 2 || len(evidence.RetryDelays) != 1 || evidence.RetryDelays[0] != 19*time.Millisecond { + t.Fatalf("attempts/delays = %#v / %#v", evidence.Attempts, evidence.RetryDelays) + } + first, second := evidence.Attempts[0], evidence.Attempts[1] + if !first.Dispatched || !first.HTTP.RequestObserved || !first.HTTP.ResponseReceived || first.HTTP.HTTPStatus == nil || *first.HTTP.HTTPStatus != http.StatusServiceUnavailable { + t.Fatalf("first attempt evidence = %+v", first) + } + if first.Failure != domain.ExecutionFailureProviderStatus || first.RequestDuration != 4*time.Millisecond { + t.Fatalf("first attempt failure/timing = %v / %v", first.Failure, first.RequestDuration) + } + if !second.Dispatched || !second.GenerationCompleted || !second.ResponseDecoded || second.HTTP.HTTPStatus == nil || *second.HTTP.HTTPStatus != http.StatusOK { + t.Fatalf("second attempt evidence = %+v", second) + } + if second.Failure != domain.ExecutionFailureNone || second.RequestDuration != 7*time.Millisecond { + t.Fatalf("second attempt failure/timing = %v / %v", second.Failure, second.RequestDuration) + } + if !second.Usage.PromptTokens.Known || !second.Usage.PromptTokens.Complete || second.Usage.PromptTokens.Value != 3 { + t.Fatalf("per-attempt usage = %+v", second.Usage.PromptTokens) + } + if !evidence.Usage.PromptTokens.Known || evidence.Usage.PromptTokens.Complete || evidence.Usage.PromptTokens.Value != 3 { + t.Fatalf("aggregate usage = %+v", evidence.Usage.PromptTokens) + } +} + +func TestGenerateWithEvidenceRetainsExhaustedAttempts(t *testing.T) { + calls := 0 + coordinator := testCoordinator(t, func(*http.Request) (*http.Response, error) { + calls++ + return response(http.StatusServiceUnavailable, `{"error":{}}`), nil + }) + coordinator.wait = func(context.Context, time.Duration) error { return nil } + + result, err, evidence := coordinator.GenerateWithEvidence(context.Background(), testRequest(1)) + var exhausted *ExhaustedError + if result != nil || !errors.As(err, &exhausted) || calls != 2 || exhausted.Attempts() != 2 { + t.Fatalf("result=%+v error=%v calls=%d", result, err, calls) + } + if len(evidence.Attempts) != 2 || len(evidence.RetryDelays) != 1 { + t.Fatalf("attempt/delay evidence = %#v / %#v", evidence.Attempts, evidence.RetryDelays) + } + for _, attempt := range evidence.Attempts { + if !attempt.Dispatched || attempt.Failure != domain.ExecutionFailureProviderStatus || attempt.HTTP.HTTPStatus == nil || *attempt.HTTP.HTTPStatus != http.StatusServiceUnavailable { + t.Fatalf("exhausted attempt evidence = %+v", attempt) + } + } + if evidence.Usage.PromptTokens.Known || evidence.Usage.PromptTokens.Complete { + t.Fatalf("missing provider usage was claimed: %+v", evidence.Usage.PromptTokens) + } +} + +func TestGenerateWithEvidenceRetainsInterruptedBackoffWithoutAttempt(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var now = time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) + calls := 0 + coordinator := testCoordinator(t, func(*http.Request) (*http.Response, error) { + calls++ + now = now.Add(3 * time.Millisecond) + return response(http.StatusServiceUnavailable, `{"error":{}}`), nil + }) + coordinator.now = func() time.Time { return now } + coordinator.wait = func(ctx context.Context, _ time.Duration) error { + now = now.Add(11 * time.Millisecond) + cancel() + return ctx.Err() + } + + result, err, evidence := coordinator.GenerateWithEvidence(ctx, testRequest(2)) + if result != nil || !errors.Is(err, context.Canceled) || calls != 1 { + t.Fatalf("result=%+v error=%v calls=%d", result, err, calls) + } + if len(evidence.Attempts) != 1 || !evidence.Attempts[0].Dispatched || len(evidence.RetryDelays) != 1 || evidence.RetryDelays[0] != 11*time.Millisecond { + t.Fatalf("interrupted evidence = %#v", evidence) + } + if evidence.Attempts[0].HTTP.HTTPStatus == nil || *evidence.Attempts[0].HTTP.HTTPStatus != http.StatusServiceUnavailable { + t.Fatalf("last dispatched response was lost: %+v", evidence.Attempts[0]) + } +} + +func TestGenerateWithEvidenceRecordsCanceledReacquisition(t *testing.T) { + var now = time.Date(2026, 10, 13, 12, 0, 0, 0, time.UTC) + calls := 0 + coordinator := testCoordinator(t, func(*http.Request) (*http.Response, error) { + calls++ + now = now.Add(2 * time.Millisecond) + return response(http.StatusServiceUnavailable, `{"error":{}}`), nil + }) + coordinator.now = func() time.Time { return now } + baseContext, cancel := context.WithCancel(context.Background()) + defer cancel() + contextWithCancelOnWait := &cancelOnDoneContext{Context: baseContext, cancel: cancel} + var releaseHeld func() + coordinator.wait = func(context.Context, time.Duration) error { + release, err := coordinator.manager.Acquire(context.Background(), "limited") + if err != nil { + t.Fatalf("hold permit: %v", err) + } + releaseHeld = release + contextWithCancelOnWait.armed.Store(true) + return nil + } + + // Signal cancellation only when the second acquisition has queued behind + // the held permit, avoiding scheduler- or sleep-based synchronization. + result, err, evidence := coordinator.GenerateWithEvidence(contextWithCancelOnWait, testRequest(2)) + if releaseHeld != nil { + releaseHeld() + } + if result != nil || !errors.Is(err, context.Canceled) || calls != 1 { + t.Fatalf("result=%+v error=%v calls=%d", result, err, calls) + } + if len(evidence.Attempts) != 2 || !evidence.Attempts[0].Dispatched || evidence.Attempts[1].Dispatched { + t.Fatalf("acquisition evidence = %#v", evidence.Attempts) + } + if evidence.Attempts[1].Failure != domain.ExecutionFailureCanceled { + t.Fatalf("canceled acquisition failure = %v", evidence.Attempts[1].Failure) + } + if evidence.Attempts[0].HTTP.HTTPStatus == nil || *evidence.Attempts[0].HTTP.HTTPStatus != http.StatusServiceUnavailable { + t.Fatalf("last dispatched evidence was lost: %+v", evidence.Attempts) + } +} + +func TestGenerateWithEvidenceNoDispatchHasCompleteZeroUsage(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + coordinator := testCoordinator(t, func(*http.Request) (*http.Response, error) { + t.Fatal("canceled generation dispatched") + return nil, nil + }) + + result, err, evidence := coordinator.GenerateWithEvidence(ctx, testRequest(0)) + if result != nil || !errors.Is(err, context.Canceled) || len(evidence.Attempts) != 0 { + t.Fatalf("result=%+v error=%v evidence=%+v", result, err, evidence) + } + if !evidence.Usage.PromptTokens.Known || !evidence.Usage.PromptTokens.Complete || evidence.Usage.PromptTokens.Value != 0 { + t.Fatalf("no-dispatch usage = %+v", evidence.Usage.PromptTokens) + } +} + +type cancelOnDoneContext struct { + context.Context + cancel context.CancelFunc + once sync.Once + armed atomic.Bool +} + +func (c *cancelOnDoneContext) Done() <-chan struct{} { + if c.armed.Load() { + c.once.Do(c.cancel) + } + return c.Context.Done() +} diff --git a/internal/observation/recorder.go b/internal/observation/recorder.go index c66ef70..3a30365 100644 --- a/internal/observation/recorder.go +++ b/internal/observation/recorder.go @@ -28,7 +28,7 @@ type Recorder struct { capacityWait time.Duration generationDuration time.Duration retryDelay time.Duration - usage usageTotals + usage UsageAccumulator } // Selection contains resolved identities only. Unresolved fields stay empty. @@ -79,8 +79,7 @@ func New(observer domain.ExecutionObserver) (*Recorder, error) { } func newRecorder(runID string, now func() time.Time, observer domain.ExecutionObserver) *Recorder { - return &Recorder{runID: runID, startedAt: now(), now: now, observer: observer, - usage: newUsageTotals()} + return &Recorder{runID: runID, startedAt: now(), now: now, observer: observer} } // NewRunID creates the identity shared by observations and successful results. @@ -119,7 +118,7 @@ func (r *Recorder) Attempt(data AttemptData) { r.lastRepairGeneration = data.Generation } r.generationDuration = addDuration(r.generationDuration, data.GenerationDuration) - r.usage.add(data.Usage, data.UsagePresence) + r.usage.Add(data.Usage, data.UsagePresence) } event := r.event(domain.ExecutionAttempt) event.Generation = data.Generation @@ -171,7 +170,7 @@ func (r *Recorder) Finish(data FinishData) { if event.Duration < 0 { event.Duration = 0 } - event.Usage = r.usage.snapshot(r.attempts) + event.Usage = r.usage.Observed() r.deliver(event) } @@ -243,7 +242,29 @@ type usageTotals struct { prompt, completion, total, cached, cacheWrite usageField } -func newUsageTotals() usageTotals { +// UsageAccumulator combines presence-aware usage across dispatched attempts. +// Its zero value represents no dispatched attempts. +type UsageAccumulator struct { + totals usageTotals + attempts uint64 +} + +// Add includes the usage reported by one dispatched attempt. +func (u *UsageAccumulator) Add(usage domain.TokenUsage, presence *domain.TokenUsagePresence) { + if u.attempts == 0 { + u.totals = completeUsageTotals() + } + u.attempts++ + u.totals.add(usage, presence) +} + +// Observed returns aggregate counts, marking every field known and complete +// when no attempt has been added. +func (u UsageAccumulator) Observed() domain.ObservedTokenUsage { + return u.totals.snapshot(u.attempts) +} + +func completeUsageTotals() usageTotals { return usageTotals{prompt: usageField{complete: true}, completion: usageField{complete: true}, total: usageField{complete: true}, cached: usageField{complete: true}, cacheWrite: usageField{complete: true}} } diff --git a/internal/observation/usage_accumulator_test.go b/internal/observation/usage_accumulator_test.go new file mode 100644 index 0000000..6046cbb --- /dev/null +++ b/internal/observation/usage_accumulator_test.go @@ -0,0 +1,27 @@ +package observation + +import ( + "testing" + + "gitea.maximumdirect.net/eric/promptkit/internal/domain" +) + +func TestUsageAccumulatorPreservesPresenceAndNoAttemptSemantics(t *testing.T) { + var empty UsageAccumulator + zero := empty.Observed() + if !zero.PromptTokens.Known || !zero.PromptTokens.Complete || zero.PromptTokens.Value != 0 || + !zero.CacheWriteTokens.Known || !zero.CacheWriteTokens.Complete || zero.CacheWriteTokens.Value != 0 { + t.Fatalf("zero-attempt usage = %+v", zero) + } + + var usage UsageAccumulator + usage.Add(domain.TokenUsage{PromptTokens: 0}, &domain.TokenUsagePresence{PromptTokens: true}) + usage.Add(domain.TokenUsage{PromptTokens: 3}, &domain.TokenUsagePresence{PromptTokens: true}) + observed := usage.Observed() + if observed.PromptTokens.Value != 3 || !observed.PromptTokens.Known || !observed.PromptTokens.Complete { + t.Fatalf("reported prompt usage = %+v", observed.PromptTokens) + } + if observed.CompletionTokens.Known || observed.CompletionTokens.Complete { + t.Fatalf("unreported completion usage = %+v", observed.CompletionTokens) + } +}