diff --git a/docs/internal/generation.md b/docs/internal/generation.md index e62ba3e..b4e3059 100644 --- a/docs/internal/generation.md +++ b/docs/internal/generation.md @@ -22,6 +22,14 @@ FIFO permit. Preparation failures dispatch nothing. When a positive retry budget ends with eligible failures, the internal exhaustion error retains the final cause and actual dispatched-attempt count. +When an invocation-local [observation recorder](observation.md) is present in +the context, each acquisition emits an attempt event, including canceled +capacity waits without dispatch. Each actual retry wait emits a delay event, +including interrupted waits. Active permits are released before callbacks; +generation duration ends before callback delivery. Built-in successful-envelope +usage retains reported-field presence, while injected clients use explicit +presence when supplied and otherwise treat zero fields as unknown. + Focused coordinator tests use controlled transport and wait functions. They cover eligibility, exhaustion, cancellation, backoff, request freezing, permit release, and the injected-client path. The capacity manager tests own permit diff --git a/docs/internal/observation.md b/docs/internal/observation.md index a715201..1862d0c 100644 --- a/docs/internal/observation.md +++ b/docs/internal/observation.md @@ -4,8 +4,9 @@ aggregation, measured phase totals, and synchronous metadata event delivery. Its event values and callback type live in `internal/domain` so internal requests can carry them without depending on the recorder package. The -recorder currently supplies run IDs to the runner; attempt and lifecycle -instrumentation is added at their owning boundaries. +recorder currently supplies run IDs to the runner and receives attempt and +retry-delay measurements from the generation coordinator. Run lifecycle +instrumentation is added at the runner boundary. A recorder keeps only counters, resolved identity fields, and the selected callback. It has no global history, queue, goroutine, or mutex held across a @@ -28,4 +29,5 @@ attempt events. The [recorder tests](../../internal/observation/recorder_test.go) own usage presence, phase sums, event ordering, empty-run accounting, identity snapshots, and callback panic behavior. The [generation coordinator](generation.md) owns -attempt and retry-delay measurements once connected to this recorder. +attempt and retry-delay measurements and emits their completion events after +releasing active permits. diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 8a46953..549a121 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -360,6 +360,8 @@ wiring in this stage. Update the internal inventory when the package lands. ## Stage 7 — Instrument Attempts And Retry Delays +**Completed:** 2026-09-23. + Connect the coordinator to the internal recorder through context. Record measured capacity wait even on interrupted acquisition, actual dispatch and outcome, usage from Stage 3, and completed/canceled backoff. Release the active permit before diff --git a/internal/capacity/manager.go b/internal/capacity/manager.go index 708e602..22379d9 100644 --- a/internal/capacity/manager.go +++ b/internal/capacity/manager.go @@ -107,6 +107,9 @@ func (m *Manager) getPool(backendID string) *pool { return m.pools[backendID] } +// Limited reports whether a backend uses an active-generation pool. +func (m *Manager) Limited(backendID string) bool { return m.getPool(backendID) != nil } + // Acquire obtains one active-generation permit for a backend. Its release // function is idempotent. Backends without a configured pool are unlimited. func (m *Manager) Acquire(ctx context.Context, backendID string) (func(), error) { diff --git a/internal/domain/domain.go b/internal/domain/domain.go index 190898f..6dd847f 100644 --- a/internal/domain/domain.go +++ b/internal/domain/domain.go @@ -284,6 +284,7 @@ type GenerateRequest struct { TargetPresence ExecutionTargetPresence StructuredOutput *StructuredOutputSpec TransientRetries uint64 + Generation uint64 } // StructuredOutputType indicates which provider-level output mode is requested. diff --git a/internal/generation/coordinator.go b/internal/generation/coordinator.go index 574a874..89d9e89 100644 --- a/internal/generation/coordinator.go +++ b/internal/generation/coordinator.go @@ -12,6 +12,7 @@ import ( "gitea.maximumdirect.net/eric/promptkit/internal/capacity" "gitea.maximumdirect.net/eric/promptkit/internal/domain" "gitea.maximumdirect.net/eric/promptkit/internal/llm" + "gitea.maximumdirect.net/eric/promptkit/internal/observation" ) const ( @@ -80,15 +81,30 @@ func (c *Coordinator) Generate(ctx context.Context, req domain.GenerateRequest) } var attempts, retriesUsed uint64 + recorder := observation.FromContext(ctx) + generation := req.Generation + if generation == 0 { + generation = 1 + } for { if err := ctx.Err(); err != nil { return nil, err } - outcome, dispatched := c.execute(ctx, req.Target.BackendID, prepared) + ordinal := attempts + 1 + outcome, dispatched, capacityWait, generationDuration := c.execute(ctx, req.Target.BackendID, prepared) if dispatched { attempts++ } + 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), + StatusCode: outcome.StatusCode, Usage: outcome.Usage, UsagePresence: outcome.UsagePresence}) + } if outcome.Err == nil { + if err := ctx.Err(); err != nil { + return nil, err + } return outcome.Response, nil } if !dispatched || !eligibleStatus(outcome) { @@ -105,8 +121,14 @@ func (c *Coordinator) Generate(ctx context.Context, req domain.GenerateRequest) } retryOrdinal := retriesUsed + 1 delay := retryDelay(c.backoff(retryOrdinal), outcome.RetryNotBefore, c.now()) - if err := c.wait(ctx, delay); err != nil { - return nil, err + waitStart := c.now() + waitErr := c.wait(ctx, delay) + if recorder != nil { + recorder.Delay(observation.DelayData{Generation: generation, Attempt: attempts, + Elapsed: elapsed(waitStart, c.now()), Failure: attemptFailure(waitErr)}) + } + if waitErr != nil { + return nil, waitErr } if err := ctx.Err(); err != nil { return nil, err @@ -116,27 +138,113 @@ func (c *Coordinator) Generate(ctx context.Context, req domain.GenerateRequest) } func (c *Coordinator) generateInjected(ctx context.Context, req domain.GenerateRequest) (*domain.GenerateResponse, error) { + recorder := observation.FromContext(ctx) + generation := req.Generation + if generation == 0 { + generation = 1 + } + start := c.now() release, err := c.manager.Acquire(ctx, req.Target.BackendID) + 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)}) + } return nil, err } defer release() 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)}) + } return nil, err } - return c.injected.Generate(ctx, req) + dispatchStart := c.now() + response, err := c.injected.Generate(ctx, req) + duration := elapsed(dispatchStart, c.now()) + if err == nil && response == nil { + err = llm.ErrMalformedResponse + } + release() + if recorder != nil { + data := observation.AttemptData{Generation: generation, Attempt: 1, Dispatched: true, CapacityWait: capacityWait, GenerationDuration: duration, + Outcome: attemptOutcome(err), Failure: attemptFailure(err)} + if err == nil { + data.Usage = response.Usage + data.UsagePresence = response.UsagePresence + } + recorder.Attempt(data) + } + if err != nil { + return nil, err + } + if err := ctx.Err(); err != nil { + return nil, err + } + return response, nil } -func (c *Coordinator) execute(ctx context.Context, backendID string, prepared *llm.PreparedGeneration) (outcome llm.AttemptOutcome, dispatched bool) { +func (c *Coordinator) execute(ctx context.Context, backendID string, prepared *llm.PreparedGeneration) (outcome llm.AttemptOutcome, dispatched bool, capacityWait, generationDuration time.Duration) { + start := c.now() release, err := c.manager.Acquire(ctx, backendID) + capacityWait = c.capacityWait(backendID, start) if err != nil { - return llm.AttemptOutcome{Err: err}, false + return llm.AttemptOutcome{Err: err}, false, capacityWait, 0 } defer release() if err := ctx.Err(); err != nil { - return llm.AttemptOutcome{Err: err}, false + return llm.AttemptOutcome{Err: err}, false, capacityWait, 0 } - return prepared.Execute(ctx), true + dispatchStart := c.now() + outcome = prepared.Execute(ctx) + return outcome, true, capacityWait, elapsed(dispatchStart, c.now()) +} + +func (c *Coordinator) capacityWait(backendID string, start time.Time) time.Duration { + if !c.manager.Limited(backendID) { + return 0 + } + return elapsed(start, c.now()) +} + +func elapsed(start, end time.Time) time.Duration { + if d := end.Sub(start); d > 0 { + return d + } + return 0 +} + +func attemptOutcome(err error) domain.ExecutionOutcome { + if err == nil { + return domain.ExecutionOutcomeSuccess + } + return domain.ExecutionOutcomeError +} + +func attemptFailure(err error) domain.ExecutionFailureKind { + if err == nil { + return domain.ExecutionFailureNone + } + if errors.Is(err, context.Canceled) { + return domain.ExecutionFailureCanceled + } + if errors.Is(err, context.DeadlineExceeded) { + return domain.ExecutionFailureDeadline + } + if errors.Is(err, llm.ErrUnexpectedStatus) { + return domain.ExecutionFailureProviderStatus + } + if errors.Is(err, llm.ErrTransport) { + return domain.ExecutionFailureTransport + } + if errors.Is(err, llm.ErrMalformedResponse) { + return domain.ExecutionFailureMalformedResponse + } + if errors.Is(err, llm.ErrInvalidRequest) { + return domain.ExecutionFailureInvalidRequest + } + return domain.ExecutionFailureGeneration } func eligibleStatus(outcome llm.AttemptOutcome) bool { diff --git a/internal/generation/observation_test.go b/internal/generation/observation_test.go new file mode 100644 index 0000000..85db1c1 --- /dev/null +++ b/internal/generation/observation_test.go @@ -0,0 +1,201 @@ +package generation + +import ( + "context" + "errors" + "net/http" + "testing" + "time" + + "gitea.maximumdirect.net/eric/promptkit/internal/capacity" + "gitea.maximumdirect.net/eric/promptkit/internal/domain" + "gitea.maximumdirect.net/eric/promptkit/internal/observation" +) + +func recordEvents(t *testing.T, ctx context.Context, events *[]domain.ExecutionEvent) context.Context { + t.Helper() + recorder, err := observation.New(func(event domain.ExecutionEvent) { *events = append(*events, event) }) + if err != nil { + t.Fatal(err) + } + return observation.WithRecorder(ctx, recorder) +} + +func TestObservedRetryAndExhaustion(t *testing.T) { + for _, status := range []int{200, 503} { + calls := 0 + coordinator := testCoordinator(t, func(*http.Request) (*http.Response, error) { + calls++ + if calls == 1 || status == 503 { + return response(503, `{"error":{}}`), nil + } + return response(200, `{"choices":[{"message":{"content":"ok"}}],"usage":{"prompt_tokens":0,"completion_tokens":2}}`), nil + }) + var events []domain.ExecutionEvent + ctx := recordEvents(t, context.Background(), &events) + _, err := coordinator.Generate(ctx, testRequest(1)) + if status == 200 && err != nil { + t.Fatal(err) + } + if status == 503 { + var exhausted *ExhaustedError + if !errors.As(err, &exhausted) { + t.Fatalf("error=%v", err) + } + } + if len(events) != 3 || events[0].Kind != domain.ExecutionAttempt || events[1].Kind != domain.ExecutionRetryDelay || events[2].Kind != domain.ExecutionAttempt { + t.Fatalf("events=%+v", events) + } + if events[0].Attempt != 1 || events[1].Attempt != 1 || events[2].Attempt != 2 || !events[0].Dispatched || !events[2].Dispatched { + t.Fatalf("indexes=%+v", events) + } + if events[0].Failure != domain.ExecutionFailureProviderStatus || events[1].RetryDelay < 0 { + t.Fatalf("first=%+v delay=%+v", events[0], events[1]) + } + if status == 200 && (!events[2].Usage.PromptTokens.Known || events[2].Usage.PromptTokens.Value != 0 || events[2].Usage.TotalTokens.Known) { + t.Fatalf("usage=%+v", events[2].Usage) + } + } +} + +func TestObservedInterruptedWaitAndCanceledCapacity(t *testing.T) { + coordinator := testCoordinator(t, func(*http.Request) (*http.Response, error) { return response(503, `{"error":{}}`), nil }) + ctx, cancel := context.WithCancel(context.Background()) + var events []domain.ExecutionEvent + ctx = recordEvents(t, ctx, &events) + coordinator.wait = func(ctx context.Context, _ time.Duration) error { cancel(); return ctx.Err() } + _, err := coordinator.Generate(ctx, testRequest(1)) + if !errors.Is(err, context.Canceled) || len(events) != 2 || events[1].Kind != domain.ExecutionRetryDelay || events[1].Failure != domain.ExecutionFailureCanceled { + t.Fatalf("error=%v events=%+v", err, events) + } + + coordinator = testCoordinator(t, func(*http.Request) (*http.Response, error) { t.Fatal("unexpected dispatch"); return nil, nil }) + release, err := coordinator.manager.Acquire(context.Background(), "limited") + if err != nil { + t.Fatal(err) + } + defer release() + ctx, cancel = context.WithCancel(context.Background()) + events = nil + ctx = recordEvents(t, ctx, &events) + entered := make(chan struct{}) + ctx = ¬ifyingContext{Context: ctx, entered: entered} + done := make(chan error, 1) + go func() { _, err := coordinator.Generate(ctx, testRequest(0)); done <- err }() + <-entered + cancel() + if err := <-done; !errors.Is(err, context.Canceled) { + t.Fatalf("error=%v", err) + } + if len(events) != 1 || events[0].Dispatched || events[0].Failure != domain.ExecutionFailureCanceled { + t.Fatalf("events=%+v", events) + } +} + +type notifyingContext struct { + context.Context + entered chan struct{} +} + +func (c *notifyingContext) Done() <-chan struct{} { + select { + case <-c.entered: + default: + close(c.entered) + } + return c.Context.Done() +} + +func TestObservedCallbackReleasesPermitAndCanCancelRetry(t *testing.T) { + coordinator := testCoordinator(t, func(*http.Request) (*http.Response, error) { return response(503, `{"error":{}}`), nil }) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + var events []domain.ExecutionEvent + recorder, err := observation.New(func(event domain.ExecutionEvent) { + events = append(events, event) + if event.Kind != domain.ExecutionAttempt { + return + } + permitCtx, stop := context.WithTimeout(context.Background(), time.Second) + defer stop() + release, err := coordinator.manager.Acquire(permitCtx, "limited") + if err != nil { + t.Errorf("callback could not acquire permit: %v", err) + } else { + release() + } + cancel() + }) + if err != nil { + t.Fatal(err) + } + _, err = coordinator.Generate(observation.WithRecorder(ctx, recorder), testRequest(2)) + if !errors.Is(err, context.Canceled) || len(events) != 1 { + t.Fatalf("error=%v events=%+v", err, events) + } +} + +type fixedClient struct { + response *domain.GenerateResponse + err error +} + +func (c fixedClient) Generate(context.Context, domain.GenerateRequest) (*domain.GenerateResponse, error) { + return c.response, c.err +} + +func TestObservedInjectedUsageAndRepairGeneration(t *testing.T) { + manager, err := capacity.NewManager(nil) + if err != nil { + t.Fatal(err) + } + var events []domain.ExecutionEvent + ctx := recordEvents(t, context.Background(), &events) + legacy := NewInjected(manager, fixedClient{response: &domain.GenerateResponse{Usage: domain.TokenUsage{PromptTokens: 3}}}) + if _, err := legacy.Generate(ctx, testRequest(4)); err != nil { + t.Fatal(err) + } + presence := &domain.TokenUsagePresence{CompletionTokens: true} + repair := NewInjected(manager, fixedClient{response: &domain.GenerateResponse{UsagePresence: presence}}) + req := testRequest(4) + req.Generation = 2 + if _, err := repair.Generate(ctx, req); err != nil { + t.Fatal(err) + } + if len(events) != 2 || !events[0].Usage.PromptTokens.Known || events[0].Usage.CompletionTokens.Known || !events[1].Usage.CompletionTokens.Known || events[1].Usage.CompletionTokens.Value != 0 || events[1].Usage.PromptTokens.Known { + t.Fatalf("events=%+v", events) + } + if events[1].Generation != 2 || events[1].Attempt != 1 { + t.Fatalf("repair indexes=%+v", events[1]) + } +} + +func TestObservedSuccessfulGenerationBeforeRepairFailure(t *testing.T) { + manager, err := capacity.NewManager(nil) + if err != nil { + t.Fatal(err) + } + var events []domain.ExecutionEvent + recorder, err := observation.New(func(event domain.ExecutionEvent) { events = append(events, event) }) + if err != nil { + t.Fatal(err) + } + ctx := observation.WithRecorder(context.Background(), recorder) + initial := NewInjected(manager, fixedClient{response: &domain.GenerateResponse{Usage: domain.TokenUsage{PromptTokens: 5}}}) + if _, err := initial.Generate(ctx, testRequest(0)); err != nil { + t.Fatal(err) + } + repair := NewInjected(manager, fixedClient{err: errors.New("repair unavailable")}) + req := testRequest(0) + req.Generation = 2 + if _, err := repair.Generate(ctx, req); err == nil { + t.Fatal("expected repair failure") + } + recorder.Finish(observation.FinishData{Outcome: domain.ExecutionOutcomeError, Failure: domain.ExecutionFailureGeneration}) + if len(events) != 3 || events[0].Outcome != domain.ExecutionOutcomeSuccess || events[1].Outcome != domain.ExecutionOutcomeError || events[2].Attempts != 2 || events[2].RepairCalls != 1 { + t.Fatalf("events=%+v", events) + } + if got := events[2].Usage.PromptTokens; got.Value != 5 || !got.Known || got.Complete { + t.Fatalf("prompt usage=%+v", got) + } +} diff --git a/internal/usecase/generation_request.go b/internal/usecase/generation_request.go index ab950c6..878c139 100644 --- a/internal/usecase/generation_request.go +++ b/internal/usecase/generation_request.go @@ -9,6 +9,7 @@ func newGenerationRequest( targetPresence domain.ExecutionTargetPresence, structuredOutput *domain.StructuredOutputSpec, transientRetries uint64, + generation uint64, ) domain.GenerateRequest { prompt.SessionID = sessionID return domain.GenerateRequest{ @@ -17,5 +18,6 @@ func newGenerationRequest( TargetPresence: targetPresence, StructuredOutput: structuredOutput, TransientRetries: transientRetries, + Generation: generation, } } diff --git a/internal/usecase/repairer.go b/internal/usecase/repairer.go index 741755e..d8d02c7 100644 --- a/internal/usecase/repairer.go +++ b/internal/usecase/repairer.go @@ -93,6 +93,7 @@ func (r *defaultOutputRepairer) Repair(ctx context.Context, req RepairRequest) ( req.TargetPresence, req.StructuredOutput, req.TransientRetries, + uint64(req.Attempt)+1, )) if err != nil { return nil, err diff --git a/internal/usecase/runner.go b/internal/usecase/runner.go index 35bfde0..3632661 100644 --- a/internal/usecase/runner.go +++ b/internal/usecase/runner.go @@ -195,6 +195,7 @@ func (r *Runner) executePreparedRun( prepared.TargetPresence, prepared.StructuredOutput, transientRetries, + 1, )) if err != nil { return nil, wrapGenerationError(err)