Expose ordinary run execution observations
This commit is contained in:
@@ -173,6 +173,34 @@ semantics. The
|
||||
[OpenAI-compatible integration contract](../integrations/openai-compatible-chat.md)
|
||||
owns the built-in client's outbound HTTP behavior.
|
||||
|
||||
### Observe An Ordinary Run
|
||||
|
||||
Set `RunRequest.Observer` to receive synchronous metadata events from `Run`:
|
||||
|
||||
```go
|
||||
request.Observer = func(event promptkit.ExecutionEvent) {
|
||||
// Translate selected metadata into an application-owned record.
|
||||
}
|
||||
result, err := engine.Run(ctx, request)
|
||||
```
|
||||
|
||||
An ordinary run emits `started` before request conversion, followed by any
|
||||
attempt and retry-delay completions, then one `finished` event on a normal
|
||||
return. `Prepare` and `PrepareExecution` emit no events. A start event has no
|
||||
resolved prompt, profile, or model; later events gain identities only as
|
||||
selection succeeds. The terminal event has dispatched-attempt, transient-retry,
|
||||
repair-call, duration, and presence-aware usage totals. Its usage is an
|
||||
aggregate, not an additional billable attempt. A known zero differs from an
|
||||
unknown count, and incomplete totals must not be treated as complete cost
|
||||
accounting. A failed operation still returns no partial `RunResult`.
|
||||
|
||||
Callbacks run synchronously, in order for one operation, and can overlap for
|
||||
different operations. Synchronize shared callback state and avoid blocking on
|
||||
work that needs the same run's admission slot. Callback panics propagate;
|
||||
there is no durable event history or promise of a terminal event after panic.
|
||||
Events omit content and error objects, but selected identifiers may still be
|
||||
sensitive. See the [observation GoDoc](../../observations.go) for exact fields.
|
||||
|
||||
### Repair A Structured Result
|
||||
|
||||
Set a small additional-call budget when a structurally invalid result can be
|
||||
|
||||
@@ -6,7 +6,7 @@ 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 and receives attempt and
|
||||
retry-delay measurements from the generation coordinator. Run lifecycle
|
||||
instrumentation is added at the runner boundary.
|
||||
instrumentation from ordinary `Run` at the facade and runner boundaries.
|
||||
|
||||
A recorder keeps only counters, resolved identity fields, and the selected
|
||||
callback. It has no global history, queue, goroutine, or mutex held across a
|
||||
|
||||
@@ -11,7 +11,7 @@ contributor workflow and validation.
|
||||
|
||||
| Component | Implemented responsibility | References |
|
||||
| --- | --- | --- |
|
||||
| Root `promptkit` package | Provides the supported engine facade, source, backend-registration, and injection options, public request, result, prompt-inspection, and profile-inspection values, opaque prepared-execution handles, profile construction, extension interfaces, value conversion, redacted formatting, and typed capacity and generation error mapping. | [Package GoDoc](../../doc.go), [prepared execution](../../prepared_execution.go), [backend API](../../backends.go), [engine facade](../../engine.go) |
|
||||
| Root `promptkit` package | Provides the supported engine facade, source, backend-registration, and injection options, public request, result, prompt-inspection, profile-inspection, and ordinary-run observation values, opaque prepared-execution handles, profile construction, extension interfaces, value conversion, redacted formatting, and typed capacity and generation error mapping. | [Package GoDoc](../../doc.go), [observations](../../observations.go), [prepared execution](../../prepared_execution.go), [backend API](../../backends.go), [engine facade](../../engine.go) |
|
||||
| `examples/go-library/prepare` | Demonstrates an offline downstream consumer using a prompt file, in-memory profile, inline input, and `Prepare`. It is not a public library package. | [Example program](../../examples/go-library/prepare/main.go) |
|
||||
| `examples/go-library/run` | Demonstrates an offline downstream consumer using a prompt file, in-memory profile, inline input, an injected deterministic model client, and `Run`. It is not a public library package. | [Example program](../../examples/go-library/run/main.go) |
|
||||
| `internal/backend` | Constructs each engine's immutable registry from maintained definitions and consumer additions, validates and defensively copies definitions through the shared JSON-value package, and consumes the LLM-owned OpenAI-compatible reserved request-field rule. | [Backend registry](../../internal/backend/registry.go) |
|
||||
@@ -29,7 +29,7 @@ contributor workflow and validation.
|
||||
| `internal/artifact` | Resolves ordinary inline and unrestricted caller-selected file references into copied artifacts with metadata and hashes. | [Internal sources and validation](sources.md) |
|
||||
| `internal/validate` | Validates basic, JSON, and JSON Schema output using operating-system filesystem or `fs.FS` schema sources and creates operation-local validation plans with canonical contained schema resources. | [Framework formats](../formats.md#schemas), [internal sources and validation](sources.md) |
|
||||
| `internal/llm` | Defines the internal generation boundary and implements outbound OpenAI-compatible chat requests from resolved execution targets, including bounded structured non-success response decoding, successful-response decoding, authentication, deadline handling, and ownership of the OpenAI-compatible reserved request-field policy. | [Internal model client](llm.md) |
|
||||
| `internal/observation` | Owns invocation-local run IDs, presence-aware usage and phase accounting, and synchronous internal metadata event delivery; generation and runner instrumentation are not yet connected. | [Internal execution observation](observation.md) |
|
||||
| `internal/observation` | Owns invocation-local run IDs, presence-aware usage and phase accounting, and synchronous metadata event delivery for ordinary runs and generation attempts. | [Internal execution observation](observation.md) |
|
||||
| `internal/usecase` | Resolves prompt definitions and hashes, profiles, backends, and targets for exact inspection and request settings for preparation, and coordinates ordinary execution and one-attempt prepared execution across internal sources, rendering, artifact loading, operation-local validation plans, generation, capacity, and bounded repair. | [Internal runner](runner.md), [prepared-execution implementation](../../internal/usecase/prepared_execution.go) |
|
||||
|
||||
The internal engine assembles these internal components without exposing their
|
||||
|
||||
@@ -134,6 +134,14 @@ and validates that artifact with the plan compiled during completion. Invalid
|
||||
generated content remains a validation result; an inability to generate or
|
||||
perform validation is an operational error.
|
||||
|
||||
When the public facade installs an invocation-local observation recorder,
|
||||
ordinary `Run` reuses its run ID and carries it to the generation coordinator
|
||||
through context. Prompt and profile identities are enriched only after their
|
||||
respective resolutions succeed, before admission. The facade starts observation
|
||||
before request conversion and emits its terminal event after public error
|
||||
mapping and after the runner's deferred admission release. Preparation and
|
||||
prepared execution have no observation delivery yet.
|
||||
|
||||
Validation preparation and execution honor cancellation at every
|
||||
Promptkit-controlled boundary and do not publish a partial plan or result.
|
||||
Schema reads are bounded and context-checked between chunks; JSON decoding,
|
||||
|
||||
@@ -382,6 +382,8 @@ Use synchronization rather than elapsed-time thresholds. Run race tests; commit.
|
||||
|
||||
## Stage 8 — Add Ordinary-Run Lifecycle And Public Observation API
|
||||
|
||||
**Completed:** 2026-09-23.
|
||||
|
||||
Add public observation types/conversion and RunRequest.Observer. The facade adapts
|
||||
the function and delegates bookkeeping to the internal recorder. Start ordinary
|
||||
Run observation before request conversion and nil-engine rejection; finish after
|
||||
|
||||
20
engine.go
20
engine.go
@@ -10,6 +10,7 @@ import (
|
||||
|
||||
"gitea.maximumdirect.net/eric/promptkit/internal/domain"
|
||||
internalengine "gitea.maximumdirect.net/eric/promptkit/internal/engine"
|
||||
"gitea.maximumdirect.net/eric/promptkit/internal/observation"
|
||||
"gitea.maximumdirect.net/eric/promptkit/internal/usecase"
|
||||
)
|
||||
|
||||
@@ -526,6 +527,25 @@ func (e *Engine) PrepareExecution(ctx context.Context, req RunRequest) (*Prepare
|
||||
// behavior. A nil Engine returns ErrInvalidConfig. Run returns no partial
|
||||
// result on error.
|
||||
func (e *Engine) Run(ctx context.Context, req RunRequest) (*RunResult, error) {
|
||||
if req.Observer == nil {
|
||||
return e.run(ctx, req)
|
||||
}
|
||||
recorder, idErr := observation.New(func(event domain.ExecutionEvent) { req.Observer(fromDomainExecutionEvent(event)) })
|
||||
recorder.Start()
|
||||
if idErr != nil {
|
||||
err := fmt.Errorf("failed to create run id: %w", idErr)
|
||||
terminal := observedFinish(nil, err)
|
||||
recorder.Finish(observation.FinishData{Outcome: terminal.Outcome, Failure: terminal.Failure})
|
||||
return nil, err
|
||||
}
|
||||
result, err := e.run(observation.WithRecorder(ctx, recorder), req)
|
||||
terminal := observedFinish(result, err)
|
||||
recorder.Finish(observation.FinishData{Outcome: terminal.Outcome, Failure: terminal.Failure,
|
||||
StatusCode: terminal.StatusCode, RetriesExhausted: terminal.RetriesExhausted})
|
||||
return result, err
|
||||
}
|
||||
|
||||
func (e *Engine) run(ctx context.Context, req RunRequest) (*RunResult, error) {
|
||||
if e == nil || e.internal == nil {
|
||||
return nil, fmt.Errorf("%w: engine is nil", ErrInvalidConfig)
|
||||
}
|
||||
|
||||
@@ -136,9 +136,16 @@ func NewRunnerWithRepairer(
|
||||
}
|
||||
|
||||
func (r *Runner) Run(ctx context.Context, req domain.RunRequest) (*domain.RunResult, error) {
|
||||
runID, err := observation.NewRunID()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create run id: %w", err)
|
||||
recorder := observation.FromContext(ctx)
|
||||
runID := ""
|
||||
if recorder != nil {
|
||||
runID = recorder.RunID()
|
||||
} else {
|
||||
var err error
|
||||
runID, err = observation.NewRunID()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to create run id: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
start := time.Now().UTC()
|
||||
@@ -324,6 +331,9 @@ func (r *Runner) resolvePreparation(
|
||||
return nil, err
|
||||
}
|
||||
def := promptSelection.definition
|
||||
if recorder := observation.FromContext(ctx); recorder != nil {
|
||||
recorder.UpdateSelection(observation.Selection{PromptID: def.ID, PromptVersion: def.Version})
|
||||
}
|
||||
promptDefinitionHash := promptSelection.hash
|
||||
effectiveContract, err := resolveOutputContract(def, req.Validation)
|
||||
if err != nil {
|
||||
@@ -342,6 +352,9 @@ func (r *Runner) resolvePreparation(
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if recorder := observation.FromContext(ctx); recorder != nil {
|
||||
recorder.UpdateSelection(observation.Selection{PromptID: def.ID, PromptVersion: def.Version, ProfileID: selection.id})
|
||||
}
|
||||
|
||||
effectiveModel, targetPresence := resolveExecutionTarget(selection.backend, selection.profile, req.Execution)
|
||||
effectiveModel.APIKey = req.APIKey
|
||||
@@ -352,6 +365,10 @@ func (r *Runner) resolvePreparation(
|
||||
if err := validateAPIKey(effectiveModel.APIKeyEnv, effectiveModel.APIKey, effectiveModel.APIKeyRequired); err != nil {
|
||||
return nil, fmt.Errorf("%w: %w", ErrInvalidRequest, err)
|
||||
}
|
||||
if recorder := observation.FromContext(ctx); recorder != nil {
|
||||
recorder.UpdateSelection(observation.Selection{PromptID: def.ID, PromptVersion: def.Version,
|
||||
ProfileID: selection.id, BackendID: effectiveModel.BackendID, Model: effectiveModel.Model})
|
||||
}
|
||||
|
||||
return &preparationState{
|
||||
definition: def,
|
||||
|
||||
233
observation_contract_test.go
Normal file
233
observation_contract_test.go
Normal file
@@ -0,0 +1,233 @@
|
||||
package promptkit_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"testing/fstest"
|
||||
|
||||
"gitea.maximumdirect.net/eric/promptkit"
|
||||
)
|
||||
|
||||
func observationEngine(t *testing.T, client promptkit.LLMClient) *promptkit.Engine {
|
||||
t.Helper()
|
||||
engine, err := promptkit.NewEngine(promptkit.Config{},
|
||||
promptkit.WithPromptFS(contractPromptFS("prompt", "profile", "hello"), "."),
|
||||
promptkit.WithProfiles(promptkit.Profile{ID: "profile", Endpoint: "https://example.invalid/v1", Model: "model"}),
|
||||
promptkit.WithLLMClient(client))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return engine
|
||||
}
|
||||
|
||||
func TestRunObservationCoversFacadeErrors(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
engine *promptkit.Engine
|
||||
request promptkit.RunRequest
|
||||
failure promptkit.ExecutionFailureKind
|
||||
}{
|
||||
{name: "nil engine", failure: promptkit.ExecutionFailureInvalidConfig},
|
||||
{name: "conversion", request: promptkit.RunRequest{AppendedMessages: []promptkit.RenderedMessage{{Role: "invalid"}}}, failure: promptkit.ExecutionFailureInvalidRequest},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
var events []promptkit.ExecutionEvent
|
||||
request := tc.request
|
||||
request.Observer = func(event promptkit.ExecutionEvent) { events = append(events, event) }
|
||||
engine := tc.engine
|
||||
if tc.name == "conversion" {
|
||||
engine = observationEngine(t, &fakeLLMClient{response: &promptkit.GenerateResponse{Content: "ok"}})
|
||||
}
|
||||
result, err := engine.Run(context.Background(), request)
|
||||
if result != nil || err == nil || len(events) != 2 || events[0].Kind != promptkit.ExecutionStarted || events[1].Kind != promptkit.ExecutionFinished || events[1].Failure != tc.failure || events[1].Attempts != 0 {
|
||||
t.Fatalf("result=%+v error=%v events=%+v", result, err, events)
|
||||
}
|
||||
if events[0].PromptID != "" || events[1].PromptID != "" || events[0].RunID == "" || events[0].RunID != events[1].RunID {
|
||||
t.Fatalf("identities=%+v", events)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunObservationHasResolvedIdentitiesAndAccounting(t *testing.T) {
|
||||
client := &fakeLLMClient{response: &promptkit.GenerateResponse{Content: "ok", Usage: promptkit.TokenUsage{PromptTokens: 3}}}
|
||||
engine := observationEngine(t, client)
|
||||
var events []promptkit.ExecutionEvent
|
||||
request := promptkit.RunRequest{PromptID: "prompt", Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }}
|
||||
result, err := engine.Run(context.Background(), request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(events) != 3 || events[0].Kind != promptkit.ExecutionStarted || events[1].Kind != promptkit.ExecutionAttempt || events[2].Kind != promptkit.ExecutionFinished {
|
||||
t.Fatalf("events=%+v", events)
|
||||
}
|
||||
if events[0].PromptID != "" || events[1].PromptID != "prompt" || events[1].PromptVersion != "1" || events[1].ProfileID != "profile" || events[1].Model != "model" {
|
||||
t.Fatalf("selection=%+v", events)
|
||||
}
|
||||
if result.RunID == "" || events[0].RunID != result.RunID || events[2].RunID != result.RunID || events[2].Attempts != 1 || events[2].TransientRetries != 0 || events[2].RepairCalls != 0 || events[2].Outcome != promptkit.ExecutionOutcomeSuccess {
|
||||
t.Fatalf("result=%+v terminal=%+v", result, events[2])
|
||||
}
|
||||
if got := events[2].Usage.PromptTokens; got.Value != 3 || !got.Known || !got.Complete {
|
||||
t.Fatalf("usage=%+v", got)
|
||||
}
|
||||
if got := events[2].Usage.CompletionTokens; got.Known || got.Complete {
|
||||
t.Fatalf("unknown usage=%+v", got)
|
||||
}
|
||||
encoded, err := json.Marshal(request)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if string(encoded) == "" || json.Valid(encoded) == false {
|
||||
t.Fatalf("request JSON=%q", encoded)
|
||||
}
|
||||
var fields map[string]json.RawMessage
|
||||
if err := json.Unmarshal(encoded, &fields); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, ok := fields["Observer"]; ok {
|
||||
t.Fatalf("observer leaked into JSON: %s", encoded)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunObservationClassifiesInvalidOutputAndGenerationFailure(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name, content string
|
||||
err error
|
||||
validation promptkit.ValidationMode
|
||||
outcome promptkit.ExecutionOutcome
|
||||
failure promptkit.ExecutionFailureKind
|
||||
}{
|
||||
{name: "invalid", content: "not json", validation: promptkit.ValidationJSON, outcome: promptkit.ExecutionOutcomeInvalidOutput},
|
||||
{name: "skipped", content: "anything", validation: promptkit.ValidationNone, outcome: promptkit.ExecutionOutcomeSuccess},
|
||||
{name: "generation error", err: errors.New("private model detail"), outcome: promptkit.ExecutionOutcomeError, failure: promptkit.ExecutionFailureGeneration},
|
||||
} {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
client := &fakeLLMClient{response: &promptkit.GenerateResponse{Content: tc.content}, err: tc.err}
|
||||
engine := observationEngine(t, client)
|
||||
var events []promptkit.ExecutionEvent
|
||||
request := promptkit.RunRequest{PromptID: "prompt", Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }}
|
||||
if tc.validation != "" {
|
||||
request.Validation = &promptkit.OutputContract{Format: promptkit.FormatText, ValidationMode: tc.validation}
|
||||
}
|
||||
result, err := engine.Run(context.Background(), request)
|
||||
if tc.err != nil && (result != nil || !errors.Is(err, promptkit.ErrLLMGenerate)) {
|
||||
t.Fatalf("result=%+v error=%v", result, err)
|
||||
}
|
||||
if tc.err == nil && (err != nil || result == nil) {
|
||||
t.Fatalf("result=%+v error=%v", result, err)
|
||||
}
|
||||
terminal := events[len(events)-1]
|
||||
if terminal.Outcome != tc.outcome || terminal.Failure != tc.failure || terminal.Attempts != 1 {
|
||||
t.Fatalf("terminal=%+v", terminal)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunObservationPromptFailureAndPanicCleanup(t *testing.T) {
|
||||
engine, err := promptkit.NewEngine(promptkit.Config{}, promptkit.WithPromptFS(fstest.MapFS{}, "."))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var events []promptkit.ExecutionEvent
|
||||
_, err = engine.Run(context.Background(), promptkit.RunRequest{PromptID: "missing", Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }})
|
||||
if !errors.Is(err, promptkit.ErrPromptNotFound) || len(events) != 2 || events[1].Failure != promptkit.ExecutionFailurePromptNotFound || events[1].PromptID != "" {
|
||||
t.Fatalf("error=%v events=%+v", err, events)
|
||||
}
|
||||
|
||||
engine = observationEngine(t, &fakeLLMClient{response: &promptkit.GenerateResponse{Content: "ok"}})
|
||||
func() {
|
||||
defer func() {
|
||||
if recover() == nil {
|
||||
t.Error("expected observer panic")
|
||||
}
|
||||
}()
|
||||
_, _ = engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", Observer: func(event promptkit.ExecutionEvent) {
|
||||
if event.Kind == promptkit.ExecutionAttempt {
|
||||
panic("callback")
|
||||
}
|
||||
}})
|
||||
}()
|
||||
result, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt"})
|
||||
if err != nil || result == nil {
|
||||
t.Fatalf("run after callback panic: result=%+v error=%v", result, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunObservationIncludesCompletedUsageBeforeRepairFailure(t *testing.T) {
|
||||
client := &fakeLLMClient{responses: []*promptkit.GenerateResponse{
|
||||
{Content: "invalid json", Usage: promptkit.TokenUsage{PromptTokens: 7}}, nil,
|
||||
}}
|
||||
engine := observationEngine(t, client)
|
||||
var events []promptkit.ExecutionEvent
|
||||
result, err := engine.Run(context.Background(), promptkit.RunRequest{
|
||||
PromptID: "prompt", Validation: &promptkit.OutputContract{Format: promptkit.FormatJSON,
|
||||
ValidationMode: promptkit.ValidationJSON, RepairAttempts: 1},
|
||||
Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) },
|
||||
})
|
||||
if result != nil || !errors.Is(err, promptkit.ErrLLMGenerate) {
|
||||
t.Fatalf("result=%+v error=%v", result, err)
|
||||
}
|
||||
if len(events) != 4 || events[1].Outcome != promptkit.ExecutionOutcomeSuccess || events[2].Generation != 2 || events[2].Outcome != promptkit.ExecutionOutcomeError {
|
||||
t.Fatalf("events=%+v", events)
|
||||
}
|
||||
terminal := events[3]
|
||||
if terminal.Attempts != 2 || terminal.RepairCalls != 1 || terminal.Failure != promptkit.ExecutionFailureMalformedResponse || terminal.Usage.PromptTokens.Value != 7 || terminal.Usage.PromptTokens.Complete {
|
||||
t.Fatalf("terminal=%+v", terminal)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunObservationValidationPreparationError(t *testing.T) {
|
||||
engine := observationEngine(t, &fakeLLMClient{response: &promptkit.GenerateResponse{Content: "ok"}})
|
||||
var events []promptkit.ExecutionEvent
|
||||
result, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", Validation: &promptkit.OutputContract{Format: promptkit.FormatJSON, ValidationMode: promptkit.ValidationJSONSchema, SchemaPath: "missing-schema.json"}, Observer: func(event promptkit.ExecutionEvent) { events = append(events, event) }})
|
||||
if result != nil || !errors.Is(err, promptkit.ErrValidation) || len(events) != 2 || events[1].Failure != promptkit.ExecutionFailureValidation || events[1].Attempts != 0 {
|
||||
t.Fatalf("result=%+v error=%v events=%+v", result, err, events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunObservationAdmissionRejection(t *testing.T) {
|
||||
artifactRelease := make(chan struct{})
|
||||
reader := &capacityArtifactReader{entered: make(chan struct{}, 2), release: artifactRelease}
|
||||
client := newCapacityGateClient(closedCapacityChannel(), 2)
|
||||
engine := newBackendCapacityEngine(t, client, 1, capacityInt(0), reader)
|
||||
first := make(chan capacityRunResult, 1)
|
||||
go runCapacityRequest(engine, context.Background(), capacityInputRequest("http://first.example/v1"), first)
|
||||
awaitCapacitySignal(t, reader.entered, "first artifact read")
|
||||
var events []promptkit.ExecutionEvent
|
||||
request := capacityInputRequest("http://second.example/v1")
|
||||
request.Observer = func(event promptkit.ExecutionEvent) { events = append(events, event) }
|
||||
result, err := engine.Run(context.Background(), request)
|
||||
close(artifactRelease)
|
||||
if firstResult := awaitCapacityRun(t, first); firstResult.err != nil {
|
||||
t.Fatalf("first run: %v", firstResult.err)
|
||||
}
|
||||
if result != nil || !errors.Is(err, promptkit.ErrCapacityExceeded) || len(events) != 2 || events[1].Failure != promptkit.ExecutionFailureCapacity || events[1].BackendID != "limited" || events[1].Attempts != 0 {
|
||||
t.Fatalf("result=%+v error=%v events=%+v", result, err, events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunObservationValuesCannotChangeExecutionAndHideDiagnostics(t *testing.T) {
|
||||
const privateDetail = "provider-private-marker"
|
||||
client := &fakeLLMClient{err: errors.New(privateDetail)}
|
||||
engine := observationEngine(t, client)
|
||||
var events []promptkit.ExecutionEvent
|
||||
_, err := engine.Run(context.Background(), promptkit.RunRequest{PromptID: "prompt", Observer: func(event promptkit.ExecutionEvent) {
|
||||
events = append(events, event)
|
||||
event.Model = "changed by callback"
|
||||
event.RunID = "changed by callback"
|
||||
}})
|
||||
if !errors.Is(err, promptkit.ErrLLMGenerate) || len(events) != 3 || events[1].RunID != events[2].RunID || events[2].Model != "model" {
|
||||
t.Fatalf("error=%v events=%+v", err, events)
|
||||
}
|
||||
encoded, err := json.Marshal(events)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if strings.Contains(string(encoded), privateDetail) || strings.Contains(string(encoded), "https://example.invalid") {
|
||||
t.Fatalf("sensitive data in events: %s", encoded)
|
||||
}
|
||||
}
|
||||
177
observations.go
Normal file
177
observations.go
Normal file
@@ -0,0 +1,177 @@
|
||||
package promptkit
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/promptkit/internal/domain"
|
||||
)
|
||||
|
||||
// ExecutionObserver receives immutable metadata events synchronously. Callers
|
||||
// must synchronize shared state if separate runs use the same observer. A
|
||||
// callback must return for its run to progress; panics propagate to the caller.
|
||||
type ExecutionObserver func(ExecutionEvent)
|
||||
|
||||
// ExecutionEventKind identifies one point in an execution's lifecycle.
|
||||
type ExecutionEventKind string
|
||||
|
||||
const (
|
||||
ExecutionStarted ExecutionEventKind = "started"
|
||||
ExecutionAttempt ExecutionEventKind = "attempt"
|
||||
ExecutionRetryDelay ExecutionEventKind = "retry_delay"
|
||||
ExecutionFinished ExecutionEventKind = "finished"
|
||||
)
|
||||
|
||||
// ExecutionOutcome reports whether the operation or attempt succeeded.
|
||||
type ExecutionOutcome string
|
||||
|
||||
const (
|
||||
ExecutionOutcomeSuccess ExecutionOutcome = "success"
|
||||
ExecutionOutcomeInvalidOutput ExecutionOutcome = "invalid_output"
|
||||
ExecutionOutcomeError ExecutionOutcome = "error"
|
||||
)
|
||||
|
||||
// ExecutionFailureKind is a content-free classification of an error.
|
||||
type ExecutionFailureKind string
|
||||
|
||||
const (
|
||||
ExecutionFailureNone ExecutionFailureKind = ""
|
||||
ExecutionFailureCanceled ExecutionFailureKind = "canceled"
|
||||
ExecutionFailureDeadline ExecutionFailureKind = "deadline"
|
||||
ExecutionFailureTransport ExecutionFailureKind = "transport"
|
||||
ExecutionFailureProviderStatus ExecutionFailureKind = "provider_status"
|
||||
ExecutionFailureMalformedResponse ExecutionFailureKind = "malformed_response"
|
||||
ExecutionFailureCapacity ExecutionFailureKind = "capacity"
|
||||
ExecutionFailureInvalidConfig ExecutionFailureKind = "invalid_config"
|
||||
ExecutionFailureInvalidRequest ExecutionFailureKind = "invalid_request"
|
||||
ExecutionFailurePromptNotFound ExecutionFailureKind = "prompt_not_found"
|
||||
ExecutionFailureProfileNotFound ExecutionFailureKind = "profile_not_found"
|
||||
ExecutionFailurePromptLoad ExecutionFailureKind = "prompt_load"
|
||||
ExecutionFailureProfileLoad ExecutionFailureKind = "profile_load"
|
||||
ExecutionFailureArtifactLoad ExecutionFailureKind = "artifact_load"
|
||||
ExecutionFailurePromptRender ExecutionFailureKind = "prompt_render"
|
||||
ExecutionFailureValidation ExecutionFailureKind = "validation"
|
||||
ExecutionFailureGeneration ExecutionFailureKind = "generation"
|
||||
ExecutionFailureOther ExecutionFailureKind = "other"
|
||||
)
|
||||
|
||||
// ObservedTokenCount distinguishes an omitted count from a reported zero.
|
||||
// Complete means every dispatched attempt reported the field.
|
||||
type ObservedTokenCount struct {
|
||||
Value int
|
||||
Known bool
|
||||
Complete bool
|
||||
}
|
||||
|
||||
// ObservedTokenUsage contains presence-aware usage for one attempt or an
|
||||
// aggregate. It does not prove complete provider billing.
|
||||
type ObservedTokenUsage struct {
|
||||
PromptTokens ObservedTokenCount
|
||||
CompletionTokens ObservedTokenCount
|
||||
TotalTokens ObservedTokenCount
|
||||
CachedTokens ObservedTokenCount
|
||||
CacheWriteTokens ObservedTokenCount
|
||||
}
|
||||
|
||||
// ExecutionEvent contains metadata only, not provider text, request content,
|
||||
// credentials, URLs, headers, or error objects. It has no stable JSON contract.
|
||||
// Identifiers themselves may be sensitive. Nonterminal aggregate counters are
|
||||
// zero; a terminal usage total must not be added to attempt usage.
|
||||
type ExecutionEvent struct {
|
||||
Kind ExecutionEventKind
|
||||
RunID string
|
||||
Time time.Time
|
||||
PromptID string
|
||||
PromptVersion string
|
||||
ProfileID string
|
||||
BackendID string
|
||||
Model string
|
||||
Generation uint64
|
||||
Attempt uint64
|
||||
Dispatched bool
|
||||
CapacityWait time.Duration
|
||||
GenerationDuration time.Duration
|
||||
RetryDelay time.Duration
|
||||
Duration time.Duration
|
||||
Outcome ExecutionOutcome
|
||||
Failure ExecutionFailureKind
|
||||
StatusCode int
|
||||
RetriesExhausted bool
|
||||
Usage ObservedTokenUsage
|
||||
Attempts uint64
|
||||
TransientRetries uint64
|
||||
RepairCalls uint64
|
||||
}
|
||||
|
||||
func fromDomainExecutionEvent(value domain.ExecutionEvent) ExecutionEvent {
|
||||
count := func(c domain.ObservedTokenCount) ObservedTokenCount {
|
||||
return ObservedTokenCount{Value: c.Value, Known: c.Known, Complete: c.Complete}
|
||||
}
|
||||
return ExecutionEvent{
|
||||
Kind: ExecutionEventKind(value.Kind), RunID: value.RunID, Time: value.Time,
|
||||
PromptID: value.PromptID, PromptVersion: value.PromptVersion, ProfileID: value.ProfileID,
|
||||
BackendID: value.BackendID, Model: value.Model, Generation: value.Generation,
|
||||
Attempt: value.Attempt, Dispatched: value.Dispatched, CapacityWait: value.CapacityWait,
|
||||
GenerationDuration: value.GenerationDuration, RetryDelay: value.RetryDelay, Duration: value.Duration,
|
||||
Outcome: ExecutionOutcome(value.Outcome), Failure: ExecutionFailureKind(value.Failure),
|
||||
StatusCode: value.StatusCode, RetriesExhausted: value.RetriesExhausted,
|
||||
Usage: ObservedTokenUsage{PromptTokens: count(value.Usage.PromptTokens), CompletionTokens: count(value.Usage.CompletionTokens),
|
||||
TotalTokens: count(value.Usage.TotalTokens), CachedTokens: count(value.Usage.CachedTokens), CacheWriteTokens: count(value.Usage.CacheWriteTokens)},
|
||||
Attempts: value.Attempts, TransientRetries: value.TransientRetries, RepairCalls: value.RepairCalls,
|
||||
}
|
||||
}
|
||||
|
||||
func observedFinish(result *RunResult, err error) domain.ExecutionEvent {
|
||||
if err == nil {
|
||||
outcome := domain.ExecutionOutcomeSuccess
|
||||
if result.Validation.Status == ValidationFailed {
|
||||
outcome = domain.ExecutionOutcomeInvalidOutput
|
||||
}
|
||||
return domain.ExecutionEvent{Outcome: outcome}
|
||||
}
|
||||
value := domain.ExecutionEvent{Outcome: domain.ExecutionOutcomeError}
|
||||
var exhausted *RetryExhaustedError
|
||||
value.RetriesExhausted = errors.As(err, &exhausted)
|
||||
var provider *GenerationError
|
||||
if errors.As(err, &provider) {
|
||||
value.StatusCode = provider.StatusCode()
|
||||
}
|
||||
switch {
|
||||
case errors.Is(err, context.Canceled):
|
||||
value.Failure = domain.ExecutionFailureCanceled
|
||||
case errors.Is(err, context.DeadlineExceeded):
|
||||
value.Failure = domain.ExecutionFailureDeadline
|
||||
case errors.Is(err, ErrTransport):
|
||||
value.Failure = domain.ExecutionFailureTransport
|
||||
case value.StatusCode != 0:
|
||||
value.Failure = domain.ExecutionFailureProviderStatus
|
||||
case errors.Is(err, ErrMalformedResponse):
|
||||
value.Failure = domain.ExecutionFailureMalformedResponse
|
||||
case errors.Is(err, ErrCapacityExceeded):
|
||||
value.Failure = domain.ExecutionFailureCapacity
|
||||
case errors.Is(err, ErrInvalidConfig):
|
||||
value.Failure = domain.ExecutionFailureInvalidConfig
|
||||
case errors.Is(err, ErrPromptNotFound):
|
||||
value.Failure = domain.ExecutionFailurePromptNotFound
|
||||
case errors.Is(err, ErrProfileNotFound):
|
||||
value.Failure = domain.ExecutionFailureProfileNotFound
|
||||
case errors.Is(err, ErrPromptLoad):
|
||||
value.Failure = domain.ExecutionFailurePromptLoad
|
||||
case errors.Is(err, ErrProfileLoad):
|
||||
value.Failure = domain.ExecutionFailureProfileLoad
|
||||
case errors.Is(err, ErrInvalidRequest):
|
||||
value.Failure = domain.ExecutionFailureInvalidRequest
|
||||
case errors.Is(err, ErrArtifactLoad):
|
||||
value.Failure = domain.ExecutionFailureArtifactLoad
|
||||
case errors.Is(err, ErrPromptRender):
|
||||
value.Failure = domain.ExecutionFailurePromptRender
|
||||
case errors.Is(err, ErrValidation):
|
||||
value.Failure = domain.ExecutionFailureValidation
|
||||
case errors.Is(err, ErrLLMGenerate):
|
||||
value.Failure = domain.ExecutionFailureGeneration
|
||||
default:
|
||||
value.Failure = domain.ExecutionFailureOther
|
||||
}
|
||||
return value
|
||||
}
|
||||
3
types.go
3
types.go
@@ -143,6 +143,9 @@ type RunRequest struct {
|
||||
// Negative values are invalid for Prepare, PrepareExecution, and Run.
|
||||
// Injected LLMClient implementations remain single-dispatch.
|
||||
TransientRetries *int
|
||||
// Observer receives metadata about this Run invocation synchronously. It is
|
||||
// excluded from JSON and is not invoked by Prepare or PrepareExecution.
|
||||
Observer ExecutionObserver `json:"-"`
|
||||
}
|
||||
|
||||
// PreparedRun contains prepared prompt execution state returned by
|
||||
|
||||
Reference in New Issue
Block a user