Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 70 additions & 0 deletions docs/llm-task-artifacts.md
Original file line number Diff line number Diff line change
Expand Up @@ -433,3 +433,73 @@ no-run lifecycle mode. Classifier failures are non-blocking: the gate warns and
continues with normal review. Successful and failed classifier task metadata
still lives under the prospective run artifact root so provider-session resume
and local artifact inspection use the same lifecycle shape.

## Request-Cost Observations (Collection Only)

Optional `request_cost_evidence` and `request_cost_checkpoint` fields supplement
schema 4 without changing output reuse or reviewer-tool coverage semantics. Only
the concrete direct OpenAI Responses adapter produces v1 request observations.
Unknown or malformed optional cost data is isolated in both metadata readers;
valid output and reviewer-tool evidence remain available. Schemas 1/2/3 retain
the existing rejection behavior. This change does not enable cost estimation.

Each observed adapter attempt, including transient retries and structured-output
corrections, gets a distinct ordinal. Usage and observed response identity are
captured independently of accepted output, including no-text, incomplete,
non-2xx and cancellation paths. Missing/null values remain unknown. Requested
and observed model/service tier are separate. Invalid counters are diagnostic
only. No prompt, raw output, refusal/error body, URL, header or credential is
copied into these observations.

The adapter-owned OpenAI client uses a fresh pinned standard HTTP transport,
disables redirects and clears the replay body. Caller-supplied clients remain
opaque. Only the actual canonical global Responses destination qualifies as
`official_global`; gateway/custom routes remain unverified. Entering `Client.Do`
means `possibly_dispatched`, including network errors; it does not prove a free
request. A pre-dispatch failure is recorded separately.

The lifecycle holds the same fail-fast advisory task lock across cache checking,
checkpoint updates, invocation and metadata, including no-run mode. Reset uses
that lock too. The bounded sidecar lives at
`llm-tasks/.request-cost/<trimmed-task-id-sha256>.json`, outside the task directory removed
by fingerprint resets and inside the same root covered by retention/purge.
Before dispatch it atomically records a pending generation. After return it
writes the final checkpoint before ledger insertion or task output/metadata.
If the pending write fails, no adapter call starts. A failed final write leaves
the on-disk pending marker as unknown. Cache loads never create generations or
ledger rows. Reset reconciles metadata observations with the sidecar before
removing metadata, retaining a known prefix even if the sidecar was lost.

Histories retain at most 64 attempts and 64 finalized generation descriptors,
16 fixed gap reasons and 256 KiB encoded checkpoints. A 4 KiB reserve keeps room
for pending/unknown state even at capacity. Prefixes, saturating omission counts
and gaps survive later success. The whole legacy metadata read is still
unbounded; only its isolated cost envelope and the sidecar reads are bounded.

Task-history completeness remains false: this slice does not establish a fresh
artifact-root accounting origin. An absent directory, empty ledger, new session
ID or successful latest request cannot establish it. Interrupted, legacy,
opaque, conflicting, missing and overflowing observations remain unknown.
Primary and repair histories remain separate detached values. Direct OpenAI
costs are not passed to the old aggregate estimator; Codex cannot borrow direct
API pricing. Existing compatibility token counters and reported-cost semantics
are not redefined as sums of the new request history.

Evidence and checkpoints have fixed-order, map-free compact JSON SHA-256
projections with explicit nulls, normalized empty arrays, sorted gap enums and
separate versioned digest domains. They bind scope, ordered attempts, generation
fingerprints, pending state and omission counts. Prices are excluded.

Two writes per generation do not preserve every pre-crash observation: a crash
during an invocation may lose that generation's already observed prefix. Its
pending marker makes the missing history explicit. The repository's atomic
writer uses rename without file/directory fsync; no power-loss durability or
provider-invoice completeness is claimed. Per-attempt durable checkpoints,
fresh-origin initialization, request-local pricing and pricing receipts require
separate implementation and verification before estimate activation.

Cost-bearing metadata uses compact JSON so its durable nested-envelope size
matches the collector/reader admission budget; unrelated legacy metadata keeps
its existing formatting. Recovered metadata-only generations contribute a known
lower-bound count, with missing finalized descriptors recorded as omissions and
gaps. Their fingerprints are not reconstructed by guessing.
46 changes: 40 additions & 6 deletions internal/llm/adapter.go
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,7 @@ type Response struct {
Usage Usage
DurationMS int64
ReviewerToolEvidence *ReviewerToolEvidence
RequestCostEvidence *RequestCostEvidence
}

// DiffToolStatus records the terminal state of the required reviewer diff tool.
Expand Down Expand Up @@ -224,9 +225,25 @@ func (e *StructuredValidationError) Is(target error) bool {
// conversation no longer exists is retried once as a fresh conversation
// instead of failing; callers that store the session for later reuse must
// persist the returned SessionID, which replaces a dangling stored one.
func RunStructuredWithSessionResume[T any](ctx context.Context, adapter Adapter, resumeSessionID string, req Request, decode Decoder[T]) (StructuredResult[T], error) {
func RunStructuredWithSessionResume[T any](ctx context.Context, adapter Adapter, resumeSessionID string, req Request, decode Decoder[T]) (result StructuredResult[T], resultErr error) {
collector := newRequestCostCollector(adapter)
defer func() {
result.Response = cloneResponse(result.Response)
result.Response.RequestCostEvidence = collector.snapshot()
// Validation error payload and returned snapshots must not share mutable
// response evidence with each other or with the adapter.
var copied []StructuredValidationAttempt
if result.ValidationAttempts != nil {
copied = make([]StructuredValidationAttempt, len(result.ValidationAttempts))
}
for i, attempt := range result.ValidationAttempts {
attempt.Response = cloneResponse(attempt.Response)
copied[i] = attempt
}
result.ValidationAttempts = copied
}()
var zero T
sessionID, response, err := runOnceWithSession(ctx, adapter, resumeSessionID, req)
sessionID, response, err := runOnceWithSession(ctx, adapter, resumeSessionID, req, collector, "initial")
if err != nil {
return StructuredResult[T]{Response: response, SessionID: sessionID}, err
}
Expand Down Expand Up @@ -255,7 +272,7 @@ func RunStructuredWithSessionResume[T any](ctx context.Context, adapter Adapter,
if strings.TrimSpace(retryResumeSessionID) == "" && !retryReq.FreshValidationRetrySession {
retryResumeSessionID = resumeSessionID
}
retrySessionID, retryResponse, err := runOnceWithSession(ctx, adapter, retryResumeSessionID, retryReq)
retrySessionID, retryResponse, err := runOnceWithSession(ctx, adapter, retryResumeSessionID, retryReq, collector, "correction")
if err != nil {
return StructuredResult[T]{Response: retryResponse, SessionID: retrySessionID, ValidationAttempts: attempts}, err
}
Expand Down Expand Up @@ -306,10 +323,15 @@ func decodeStructuredAccepted[T any](decode Decoder[T], data []byte) (T, []byte,
// prior context, neither of which the review depends on. Without the fallback a
// dangling session ID fails the task outright, and a whole round of reviewers
// can fail before doing any work while still reporting zero findings.
func runOnceWithSession(ctx context.Context, adapter Adapter, resumeSessionID string, req Request) (string, Response, error) {
func runOnceWithSession(ctx context.Context, adapter Adapter, resumeSessionID string, req Request, collector *requestCostCollector, phase string) (string, Response, error) {
var adapterAttempt uint32
sessionDropped := false
for attempt := 0; ; attempt++ {
if adapterAttempt < ^uint32(0) {
adapterAttempt++
}
sid, resp, err := runOnceAttempt(ctx, adapter, resumeSessionID, req)
collector.observe(resp, phase, adapterAttempt)
if err == nil {
return sid, resp, nil
}
Expand Down Expand Up @@ -339,15 +361,19 @@ func runOnceAttempt(ctx context.Context, adapter Adapter, resumeSessionID string
)
if req.ReviewerWorkspace != nil {
if err := RequireReviewerWorkspace(adapter); err != nil {
return "", Response{}, err
return "", nonDispatchedCostResponse(adapter, req), err
}
}
if strings.TrimSpace(resumeSessionID) != "" && adapter.SupportsResume() {
resuming := strings.TrimSpace(resumeSessionID) != "" && adapter.SupportsResume()
if resuming {
stream, err = adapter.Resume(ctx, resumeSessionID, req)
} else {
stream, err = adapter.Start(ctx, req)
}
if err != nil {
if stream == nil && !resuming {
return "", nonDispatchedCostResponse(adapter, req), err
}
return "", Response{}, err
}
response, err := stream.Wait(ctx)
Expand All @@ -356,7 +382,15 @@ func runOnceAttempt(ctx context.Context, adapter Adapter, resumeSessionID string

func cloneResponse(response Response) Response {
response.StructuredOutput = append([]byte(nil), response.StructuredOutput...)
response.Usage.TokensIn = costCopy(response.Usage.TokensIn)
response.Usage.TokensOut = costCopy(response.Usage.TokensOut)
response.Usage.CacheRead = costCopy(response.Usage.CacheRead)
response.Usage.CacheCreate = costCopy(response.Usage.CacheCreate)
response.Usage.CacheCreate5m = costCopy(response.Usage.CacheCreate5m)
response.Usage.CacheCreate1h = costCopy(response.Usage.CacheCreate1h)
response.Usage.CostUSD = costCopy(response.Usage.CostUSD)
response.ReviewerToolEvidence = CloneReviewerToolEvidence(response.ReviewerToolEvidence)
response.RequestCostEvidence = CloneRequestCostEvidence(response.RequestCostEvidence)
return response
}

Expand Down
Loading
Loading