Document and demonstrate execution controls
This commit is contained in:
@@ -581,6 +581,59 @@ retained callback without invoking it. The function's closure is retained by
|
||||
identity, so synchronize any shared state it uses. Prepared start events can
|
||||
include identities already resolved in the frozen snapshot.
|
||||
|
||||
## Bound Execution And Retain Prepared Details
|
||||
|
||||
Select the optional controls on [RunRequest](../../types.go). The operation
|
||||
budget is separate from the provider timeout and cancellation is cooperative.
|
||||
Notification is conditional on reaching the prepared point, synchronous, and
|
||||
runs while admission is held; its caller-owned details can be retained even if
|
||||
generation later fails. Do not rely on notification for preparation or admission
|
||||
failures. The application owns any subsequent diagnostic handling.
|
||||
|
||||
```go
|
||||
var details *promptkit.PreparedRun
|
||||
result, err := engine.Run(ctx, promptkit.RunRequest{
|
||||
PromptID: "answer", OperationTimeout: 30 * time.Second,
|
||||
OnPrepared: func(_ context.Context, prepared promptkit.PreparedRun) { details = &prepared },
|
||||
})
|
||||
if errors.Is(err, promptkit.ErrOperationTimeout) && errors.Is(err, context.DeadlineExceeded) {
|
||||
// details is available only if the notification was reached.
|
||||
}
|
||||
```
|
||||
|
||||
For coordinated turns, put the controls on each returned request. Assessment
|
||||
in the next-turn callback runs after the previous admission and child scope
|
||||
are released, so it can select its own budget:
|
||||
|
||||
```go
|
||||
next := func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
|
||||
if turn.Result != nil {
|
||||
_, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "assessment", OperationTimeout: 10 * time.Second})
|
||||
return nil, err
|
||||
}
|
||||
return &promptkit.RunRequest{PromptID: "answer", OperationTimeout: 30 * time.Second,
|
||||
OnPrepared: func(_ context.Context, prepared promptkit.PreparedRun) { details = &prepared }}, nil
|
||||
}
|
||||
```
|
||||
|
||||
For split execution, [PrepareExecution](../../engine.go) retains the controls
|
||||
without starting their timer; idle time does not count. Use the same caller
|
||||
deadline for both calls if preparation and execution need one shared bound.
|
||||
|
||||
```go
|
||||
handle, err := engine.PrepareExecution(ctx, promptkit.RunRequest{
|
||||
PromptID: "answer", OperationTimeout: 30 * time.Second,
|
||||
OnPrepared: func(_ context.Context, prepared promptkit.PreparedRun) { details = &prepared },
|
||||
})
|
||||
if err != nil { return err }
|
||||
defer handle.Discard()
|
||||
result, err := engine.RunPrepared(ctx, handle)
|
||||
```
|
||||
|
||||
See the [observation example](../../examples/go-library/observe/main.go) for an
|
||||
offline coordinated workflow and [ErrOperationTimeout](../../engine.go) for
|
||||
timeout attribution.
|
||||
|
||||
## Execute And Validate
|
||||
|
||||
[`Engine.Run`](../../engine.go) performs the same preparation, invokes the
|
||||
|
||||
@@ -14,6 +14,9 @@ The public projection retains private operation-owner metadata separately from
|
||||
inspectable child causes. Terminal classification uses that owner, preserving
|
||||
context precedence and generation-specific transport/provider details rather
|
||||
than mistaking a collaborator's nested lookup error for the parent's operation.
|
||||
Timeout attribution is finalized and the child scope released before terminal
|
||||
delivery, so observer latency cannot change the already selected outcome.
|
||||
Prepared-content notification is a separate opt-in callback, not an event.
|
||||
|
||||
The facade masks inherited recorder and prepared-claim state when entering
|
||||
preparation or execution. Delegated calls from an injected collaborator retain
|
||||
|
||||
@@ -14,7 +14,7 @@ contributor workflow and validation.
|
||||
| Root `promptkit` package | Provides the supported engine facade, ConfigKit source helpers, collection and resource mapping conveniences, injection options, public requests and results, resource and requirement inspection and identity inventories, concrete-request validation, explicit smoke diagnostics, observation values, opaque prepared-execution handles, 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), [smoke diagnostics](../../smoke_diagnostics.go), [identity listing](../../listing.go), [requirement facade](../../requirements.go) |
|
||||
| `examples/go-library/prepare` | Demonstrates offline preparation from two explicitly selected synthetic collections, source-relative imported content, a scoped view, filtered listings, and operator inventory. It is not a public library package. | [Example program](../../examples/go-library/prepare/main.go), [collection inputs](../../examples/go-library/prepare/collections/support/collection.yaml) |
|
||||
| `examples/go-library/run` | Demonstrates an offline downstream consumer using a prompt file, programmatic 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) |
|
||||
| `examples/go-library/observe` | Demonstrates offline built-in HTTP retry and synchronous metadata observation across structured proposal and assessment runs using embedded sources, two profiles on one finite backend, and an application decision. It is not a public library package. | [Example program](../../examples/go-library/observe/main.go) |
|
||||
| `examples/go-library/observe` | Demonstrates offline retry, observation, prepared notification, and separately budgeted coordinated proposal and assessment runs using two profiles on one finite backend. It is not a public library package. | [Example program](../../examples/go-library/observe/main.go) |
|
||||
| `internal/backend` | Holds a defensive, immutable lookup of final backend definitions, including diagnostic metadata, plus capacity policies validated by the configuration adapter. | [Backend registry](../../internal/backend/registry.go) |
|
||||
| `internal/configuration` | Loads root and namespaced resources through ConfigKit; validates final values, captures indexed resource views and source-relative prompt content, compiled templates, requirement metadata, identity inventories, backend and presence-aware execution profiles, and the read-only provenance inspection view. | [Typed loader](../../internal/configuration/load.go), [identity listing](../../internal/configuration/listing.go), [internal sources](sources.md#typed-configuration-loading) |
|
||||
| `internal/capacity` | Owns engine-local bounded execution admission and FIFO model-generation permits for limited backend IDs, including cancellation-safe waiter removal and client wrapping. | [Internal capacity management](capacity.md) |
|
||||
|
||||
@@ -209,6 +209,16 @@ mapping and after the runner's deferred admission release. Preparation and
|
||||
prepared execution follows the same terminal mapping after a successful claim;
|
||||
invalid or discarded handles do not acquire an observer.
|
||||
|
||||
The facade establishes an optional operation scope for ordinary runs before
|
||||
observation and preparation. A claimed prepared execution establishes its own
|
||||
scope from the retained duration; preparation and idle time do not consume it.
|
||||
The shared scope finalizer marks only a context failure caused by that scope's
|
||||
own deadline, preserving its underlying operation and cause. The scope is
|
||||
released before terminal delivery. At the shared prepared execution boundary,
|
||||
the runner conditionally invokes the adapter for the actual prepared snapshot
|
||||
while admission is held, with context checks before and after notification.
|
||||
The facade adapter projects a detached public value only when selected.
|
||||
|
||||
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,
|
||||
|
||||
@@ -398,6 +398,8 @@ be redesigned solely because its ordinary run gained options.
|
||||
|
||||
## Stage 8: Document, Demonstrate, And Validate The Assembled Feature
|
||||
|
||||
Status: complete.
|
||||
|
||||
Outcome: the implementation and its supported consumer workflow are complete,
|
||||
with documentation in the canonical owners.
|
||||
|
||||
|
||||
13
engine.go
13
engine.go
@@ -63,6 +63,9 @@ var (
|
||||
// ErrOperationTimeout identifies expiration of the timeout selected by
|
||||
// RunRequest.OperationTimeout. Such errors also match context.DeadlineExceeded
|
||||
// and retain their underlying operation and collaborator identities.
|
||||
// Earlier parent cancellation or deadlines and provider-only deadlines do
|
||||
// not acquire this identity. A completed result remains completed even if
|
||||
// a terminal observer outlasts the selected budget.
|
||||
ErrOperationTimeout = errors.New("operation timeout exceeded")
|
||||
// ErrLLMGenerate identifies a model-client failure or a nil successful
|
||||
// response. A built-in OpenAI-compatible non-2xx response is available as a
|
||||
@@ -390,6 +393,7 @@ func (e *Engine) InspectBackend(ctx context.Context, backendID string) (*Backend
|
||||
// reported in the applicable operation category; no general errors.Is
|
||||
// relationship to ctx.Err is promised. Prepare returns no partial result on
|
||||
// error. It validates RunRequest.TransientRetries but makes no model attempts.
|
||||
// RunRequest.OperationTimeout and RunRequest.OnPrepared have no effect here.
|
||||
func (e *Engine) Prepare(ctx context.Context, req RunRequest) (*PreparedRun, error) {
|
||||
prepared, err := e.prepareRequest(ctx, req)
|
||||
if err != nil {
|
||||
@@ -411,6 +415,7 @@ func (e *Engine) Prepare(ctx context.Context, req RunRequest) (*PreparedRun, err
|
||||
// matching ErrInvalidConfig when the Engine is nil. Success does not guarantee
|
||||
// provider acceptance or later execution success. Use [Engine.PrepareExecution]
|
||||
// when execution must use the snapshot produced by preparation.
|
||||
// Execution-only controls do not start a timer or invoke a notification here.
|
||||
func (e *Engine) ValidateRequest(ctx context.Context, req RunRequest) error {
|
||||
_, err := e.prepareRequest(ctx, req)
|
||||
return err
|
||||
@@ -453,6 +458,8 @@ func (e *Engine) prepareRequest(ctx context.Context, req RunRequest) (*domain.Pr
|
||||
// ErrInvalidConfig. The effective transient retry budget is frozen privately
|
||||
// for RunPrepared and is absent from Details. A supplied Observer is retained
|
||||
// privately for RunPrepared; preparation itself emits no events.
|
||||
// RunRequest.OperationTimeout and RunRequest.OnPrepared are retained without
|
||||
// starting a timer or invoking the notification during preparation.
|
||||
func (e *Engine) PrepareExecution(ctx context.Context, req RunRequest) (*PreparedExecution, error) {
|
||||
ctx = observation.WithoutExecutionState(ctx)
|
||||
if e == nil || e.internal == nil {
|
||||
@@ -483,6 +490,9 @@ func (e *Engine) PrepareExecution(ctx context.Context, req RunRequest) (*Prepare
|
||||
// returns the final failed validation result with cumulative usage and actual
|
||||
// repair attempts. An inability to generate or validate returns an error and
|
||||
// no partial result.
|
||||
// RunRequest.OperationTimeout selects a cooperative whole-operation deadline;
|
||||
// RunRequest.OnPrepared selects conditional notification before generation.
|
||||
// The outcome and timeout attribution are fixed before terminal observation.
|
||||
//
|
||||
// The built-in client retries eligible provider statuses using the effective
|
||||
// Config or request budget. With R additional transient attempts and P repair
|
||||
@@ -600,6 +610,9 @@ func (e *Engine) run(ctx context.Context, req RunRequest) (*RunResult, error) {
|
||||
// mapping. Invalid, foreign, discarded, or reused handles emit no events.
|
||||
// Callbacks run outside the handle and capacity locks; panics propagate and
|
||||
// do not promise a terminal event.
|
||||
// The retained RunRequest.OperationTimeout starts after claim; the retained
|
||||
// RunRequest.OnPrepared is notified only if execution reaches its prepared
|
||||
// point. Final timeout attribution precedes terminal observation.
|
||||
func (e *Engine) RunPrepared(ctx context.Context, prepared *PreparedExecution) (*RunResult, error) {
|
||||
ctx = observation.WithoutExecutionState(ctx)
|
||||
if e == nil || e.internal == nil {
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"go.maximumdirect.net/configkit"
|
||||
"go.maximumdirect.net/promptkit"
|
||||
@@ -79,6 +80,8 @@ type summary struct {
|
||||
ExhaustedOutcome promptkit.CoordinationOutcome `json:"exhausted_outcome"`
|
||||
GroupMembers int `json:"group_members"`
|
||||
GroupAttempts uint64 `json:"group_attempts"`
|
||||
PreparedMessages int `json:"prepared_messages"`
|
||||
PreparedParentMatches bool `json:"prepared_parent_matches"`
|
||||
}
|
||||
|
||||
func run() (summary, error) {
|
||||
@@ -123,17 +126,18 @@ func run() (summary, error) {
|
||||
return summary{}, err
|
||||
}
|
||||
var assessmentLog eventLog
|
||||
var preparedDetails promptkit.PreparedRun
|
||||
decision := "rejected"
|
||||
coordination, err := engine.Coordinate(context.Background(), promptkit.CoordinateRequest{
|
||||
Start: proposal.Exchange, AdditionalTurns: 1, Observer: executions.Observe,
|
||||
Next: func(ctx context.Context, turn promptkit.ConversationTurn) (*promptkit.RunRequest, error) {
|
||||
if turn.Result == nil {
|
||||
return &promptkit.RunRequest{PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Refine the proposal.")}}, nil
|
||||
return &promptkit.RunRequest{PromptID: "example.proposal", Inputs: map[string]promptkit.ArtifactRef{"idea": promptkit.Inline("Refine the proposal.")}, OperationTimeout: 10 * time.Second, OnPrepared: func(_ context.Context, details promptkit.PreparedRun) { preparedDetails = details }}, nil
|
||||
}
|
||||
if !turn.Result.Validation.IsValid {
|
||||
return nil, fmt.Errorf("follow-up failed structural validation")
|
||||
}
|
||||
assessment, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "example.assessment", Inputs: map[string]promptkit.ArtifactRef{"proposal": promptkit.Inline(turn.Result.RawOutput)}, Observer: promptkit.ComposeObservers(assessmentLog.record, executions.Observe)})
|
||||
assessment, err := engine.Run(ctx, promptkit.RunRequest{PromptID: "example.assessment", Inputs: map[string]promptkit.ArtifactRef{"proposal": promptkit.Inline(turn.Result.RawOutput)}, OperationTimeout: 10 * time.Second, Observer: promptkit.ComposeObservers(assessmentLog.record, executions.Observe)})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -183,6 +187,8 @@ func run() (summary, error) {
|
||||
ExhaustedOutcome: exhausted.Outcome,
|
||||
GroupMembers: len(group.Members),
|
||||
GroupAttempts: group.Attempts,
|
||||
PreparedMessages: len(preparedDetails.Messages),
|
||||
PreparedParentMatches: preparedDetails.ParentRunID == proposal.RunID,
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -12,23 +12,25 @@ func TestObserveExample(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var got struct {
|
||||
ProposalAttempts *int `json:"proposal_attempts"`
|
||||
ProposalRetries *int `json:"proposal_transient_retries"`
|
||||
UsageComplete *bool `json:"proposal_usage_complete"`
|
||||
AssessmentAttempts *int `json:"assessment_attempts"`
|
||||
AssessmentRetries *int `json:"assessment_transient_retries"`
|
||||
Outcome string `json:"assessment_outcome"`
|
||||
CoordinationOutcome string `json:"coordination_outcome"`
|
||||
ExhaustedOutcome string `json:"exhausted_outcome"`
|
||||
GroupMembers int `json:"group_members"`
|
||||
GroupAttempts int `json:"group_attempts"`
|
||||
ProposalAttempts *int `json:"proposal_attempts"`
|
||||
ProposalRetries *int `json:"proposal_transient_retries"`
|
||||
UsageComplete *bool `json:"proposal_usage_complete"`
|
||||
AssessmentAttempts *int `json:"assessment_attempts"`
|
||||
AssessmentRetries *int `json:"assessment_transient_retries"`
|
||||
Outcome string `json:"assessment_outcome"`
|
||||
CoordinationOutcome string `json:"coordination_outcome"`
|
||||
ExhaustedOutcome string `json:"exhausted_outcome"`
|
||||
GroupMembers int `json:"group_members"`
|
||||
GroupAttempts int `json:"group_attempts"`
|
||||
PreparedMessages int `json:"prepared_messages"`
|
||||
PreparedParentMatches bool `json:"prepared_parent_matches"`
|
||||
}
|
||||
if err := json.Unmarshal(output.Bytes(), &got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if got.ProposalAttempts == nil || *got.ProposalAttempts != 2 || got.ProposalRetries == nil || *got.ProposalRetries != 1 ||
|
||||
got.UsageComplete == nil || *got.UsageComplete || got.AssessmentAttempts == nil || *got.AssessmentAttempts != 1 ||
|
||||
got.AssessmentRetries == nil || *got.AssessmentRetries != 0 || got.Outcome != "accepted" || got.CoordinationOutcome != "stopped" || got.ExhaustedOutcome != "exhausted" || got.GroupMembers != 3 || got.GroupAttempts != 4 {
|
||||
got.AssessmentRetries == nil || *got.AssessmentRetries != 0 || got.Outcome != "accepted" || got.CoordinationOutcome != "stopped" || got.ExhaustedOutcome != "exhausted" || got.GroupMembers != 3 || got.GroupAttempts != 4 || got.PreparedMessages < 3 || !got.PreparedParentMatches {
|
||||
t.Fatalf("unexpected observation summary: %+v", got)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -79,3 +79,22 @@ func TestPreparedExecutionRespectsCallerDeadline(t *testing.T) {
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestPreparedExecutionDoesNotTimePreparation(t *testing.T) {
|
||||
synctest.Test(t, func(t *testing.T) {
|
||||
release := make(chan struct{})
|
||||
reader := &capacityArtifactReader{entered: make(chan struct{}, 1), release: release}
|
||||
engine := newBackendCapacityEngine(t, newCapacityGateClient(closedCapacityChannel(), 1), 1, capacityInt(0), reader)
|
||||
req := capacityInputRequest("http://example.invalid/v1")
|
||||
req.OperationTimeout = time.Second
|
||||
go func() { time.Sleep(2 * time.Second); close(release) }()
|
||||
start := time.Now()
|
||||
handle, err := engine.PrepareExecution(context.Background(), req)
|
||||
if err != nil || time.Since(start) != 2*time.Second {
|
||||
t.Fatalf("prepare elapsed=%v error=%v", time.Since(start), err)
|
||||
}
|
||||
if _, err = engine.RunPrepared(context.Background(), handle); err != nil {
|
||||
t.Fatalf("execution after preparation: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -18,6 +18,9 @@ const preparedExecutionString = "promptkit.PreparedExecution{opaque}"
|
||||
// It also retains RunRequest.CaptureExchange and RunRequest.ContinueFrom for
|
||||
// execution of the frozen snapshot. Only RunPrepared can produce its completed
|
||||
// [Exchange]; preparation alone does not.
|
||||
// RunRequest.OperationTimeout and RunRequest.OnPrepared are retained as
|
||||
// selections without an active idle timer; their execution lifecycle is
|
||||
// described on those fields.
|
||||
type PreparedExecution struct {
|
||||
internal *usecase.PreparedExecution
|
||||
}
|
||||
|
||||
8
types.go
8
types.go
@@ -89,7 +89,10 @@ type RunRequest struct {
|
||||
// applies it across preparation and execution; PrepareExecution retains it
|
||||
// for RunPrepared, without counting preparation or idle time. Zero or a
|
||||
// negative value selects no additional timeout. This is independent of the
|
||||
// provider generation timeout. Cancellation remains cooperative.
|
||||
// provider generation timeout and respects earlier parent cancellation or
|
||||
// deadlines. Local timeout failures match ErrOperationTimeout and
|
||||
// context.DeadlineExceeded. Cancellation remains cooperative; callbacks or
|
||||
// collaborators that ignore context cannot be forcibly interrupted.
|
||||
OperationTimeout time.Duration `json:"-" yaml:"-"`
|
||||
// OnPrepared synchronously receives detached, credential-redacted details of
|
||||
// the logical request after preparation and admission, before generation,
|
||||
@@ -182,7 +185,8 @@ type RunRequest struct {
|
||||
|
||||
// PreparedRun contains prepared prompt execution state returned by
|
||||
// [Engine.Prepare] or [PreparedExecution.Details]. It does not include resolved
|
||||
// API key values, model output, validation results, or internal target presence
|
||||
// API key values. RunRequest.OnPrepared also supplies this detached value.
|
||||
// It does not include model output, validation results, or internal target presence
|
||||
// metadata. PreparedRun has a stable JSON representation.
|
||||
//
|
||||
// All maps, slices, pointers, and schema values are caller-owned copies. JSON
|
||||
|
||||
Reference in New Issue
Block a user