Telemetry updates

Author: santoshkumarradha

docs/TELEMETRY.md

@@ -51,6 +51,10 @@ any of the three drift apart.
 | usage_delta | input_tokens | provider-reported input tokens since the preceding usage event |
 | usage_delta | output_tokens | provider-reported output tokens since the preceding usage event |
 | usage_delta | total_tokens | the sum of this event's input and output tokens |
+| usage_delta | routing_provider | openrouter, openai, anthropic, google, bedrock, azure, ollama, other, or unknown |
+| usage_delta | model_family | deepseek, claude, gpt, gemini, llama, qwen, kimi, glm, other, or unknown |
+| usage_delta | usage_status | reported or missing |
+| usage_delta | accounting_source | provider or engine |
 | session_ended | mode | chat or task |
 | session_ended | duration | a band: under 1m, 1-5m, 5-30m, 30m-2h, 2h or more |
 | session_ended | turns | a count band |
@@ -67,8 +71,8 @@ any of the three drift apart.
 
 Count bands are 0, 1, 2-5, 6-20, 21-100 and 100+. Dollar bands are 0, under
 0.01, 0.01-0.1, 0.1-1, 1-10 and 10+.
-Each numeric usage delta is written when provider accounting arrives, including
-tokens read from cache once. Missing usage contributes no event. Deltas, rather
+Each provider usage receipt is appended locally before provider accounting returns, including
+tokens read from cache once. Missing provider receipts produce a `usage_status = missing` event without token fields; a missing receipt never invents a zero total. Routing categories follow the actual attempt, including an explicitly authorized billing-door switch and any later generation receipt. Bounded routing and model-family categories describe reported usage without sending exact model names or endpoint addresses. Hosted engines currently report aggregate turn deltas with unknown routing and family, labelled `accounting_source = engine`; those aggregates are not per-call receipts and can be lost before a turn ends. Deltas, rather
 than repeated session totals, make the stream safe to sum while a run is open.
 
 first_run carries only the every-event properties and is sent once per
@@ -82,7 +86,7 @@ ids never leave this machine.
 
 Prompts, model replies, code, file names, paths, repo or directory names, git
 remotes, hostnames, usernames, IP addresses, environment values, API keys,
-email addresses, model names, error text, panic messages. A test builds every
+email addresses, exact model names, custom model names, error text, panic messages. Public model-family categories are sent as described above. A test builds every
 event from inputs stuffed with exactly these and fails if any of them reach
 the marshalled output.
 
@@ -93,7 +97,7 @@ per request, every 30 seconds while a session is open and once more when it
 ends. Nothing older than 7 days is sent and at most 1000 lines are kept. An
 event whose version is unknown is dropped at send time and never leaves the
 machine. The spool is plain JSON lines, so what has not left yet can be read
-with any editor.
+with any editor. A local warning reports storage failures or counts discarded by the age/size limits; it contains no event contents. These warnings do not establish how many tokens were lost.
 
 ## Turning it off
 
@@ -148,3 +152,5 @@ the count GitHub already shows on the release page. Sidecar files such as
 `.github/workflows/release-downloads.yml`; the script behind it is
 `scripts/release_downloads.py`, and running it with `--dry-run` prints the
 batch it would send.
+
+The release-download workflow fails visibly when its production PostHog key or GitHub token is missing; dry-run still works without a PostHog key.

docs/changes/unreleased/1755-telemetry-diagnostics.md

@@ -0,0 +1,11 @@
+---
+kind: changed
+title: Usage receipts distinguish routing, model families and missing counts
+pr: 1755
+surface: [engine, remote, build, docs]
+invalidates:
+  - "Usage telemetry carried only positive token totals. It now includes bounded routing and model-family categories and explicitly records missing provider receipts without token totals."
+  - "Provider usage appends ran on background goroutines. Completed receipts are now appended before accounting returns, while network delivery stays periodic."
+  - "Usage categories named the initial service even after a plan overflow. They now follow the actual attempt route, including later generation receipt recovery. Periodic delivery no longer holds the local append lock across network requests."
+  - "Missing release-download credentials silently passed the scheduled job. Production reporting now fails visibly when required credentials are absent."
+---

internal/manual/chat/telemetry.md

@@ -0,0 +1,9 @@
+# Usage counts and privacy
+
+## What anonymous usage counts collect
+
+codeaf sends anonymous session counts and provider-reported input/output token counts when telemetry is enabled. Usage receipts include a bounded routing-service category and public model family, never an exact model name, endpoint address, prompt, code, path or API key. Missing provider receipts have a missing status and no invented token total. Completed local provider receipts are appended before accounting returns, then sent every 30 seconds and at shutdown. Hosted chat currently sends turn aggregates with unknown provider and family; work lost before a turn ends is not guaranteed to appear.
+
+## Turning anonymous counts off
+
+Set `CODEAF_TELEMETRY=off`, `DO_NOT_TRACK=1`, or the telemetry switch in `/settings`. Project settings can also turn telemetry off. Disabled telemetry sends no usage receipts. The detailed public contract is in `docs/TELEMETRY.md`.

internal/provider/calllog.go

@@ -121,6 +121,9 @@ type callTrace struct {
 	// row written after the transport returned always names the attempt that
 	// actually produced the answer.
 	attemptID string
+	// attemptBase is the address this attempt actually used, including an
+	// explicitly authorized plan overflow. It never leaves the process.
+	attemptBase string
 	// open is the attempt that has a start row on the wire and nothing under it
 	// yet, and nil whenever the log is square.
 	//
@@ -171,6 +174,7 @@ type openAttempt struct {
 	request *ai.Request
 	knobs   callKnobs
 	stream  bool
+	base    string
 }
 
 // The words [calllog.Record.Ended] uses. They are constants for the reason the
@@ -212,6 +216,7 @@ func (t *callTrace) track(facts recordFacts, id string) {
 		request: facts.request,
 		knobs:   facts.knobs,
 		stream:  facts.stream,
+		base:    facts.baseFor(t.attemptBase),
 	}
 }
 
@@ -229,7 +234,7 @@ func (c *Client) closeOpenAttempt(ctx context.Context, trace *callTrace, ended s
 	trace.open = nil
 	c.record(recordFacts{
 		ctx: ctx, request: open.request, knobs: open.knobs, stream: open.stream,
-		attempt: open.attempt, began: open.began, ended: ended, id: open.id,
+		attempt: open.attempt, began: open.began, ended: ended, id: open.id, base: open.base,
 	})
 }
 
@@ -270,6 +275,9 @@ func (t *callTrace) note(learned ...string) {
 // endpoint has no status and no tokens, and the emptiness law leaves both off
 // the line rather than writing a zero somebody could read as a measurement.
 type recordFacts struct {
+	// base is explicit on synthetic endings of an earlier attempt.
+	base string
+
 	ctx     context.Context
 	request *ai.Request
 	knobs   callKnobs
@@ -519,7 +527,15 @@ func (c *Client) record(facts recordFacts) {
 	// sentence is the failure verdict and its cost the spend, 0 when absent.
 	if record.Phase != calllog.PhaseStart {
 		telemetry.CountModelCall(record.Error == "", record.Cost)
-		telemetry.CountTokens(record.PromptTokens, record.CompletionTokens)
+		status := "missing"
+		if facts.response != nil && facts.response.Usage != nil {
+			status = "reported"
+		}
+		telemetry.CountUsage(record.PromptTokens, record.CompletionTokens, telemetry.UsageDimensions{
+			RoutingProvider: telemetry.RoutingProvider(facts.baseFor(c.config.BaseURL)),
+			ModelFamily:     telemetry.ModelFamily(model), UsageStatus: status,
+			AccountingSource: "provider", ReceiptID: record.ID,
+		})
 	}
 	c.recordBodies(facts, record, model)
 }
@@ -820,3 +836,15 @@ func (c *Client) relaxNames(model string, knobs callKnobs) []string {
 	}
 	return names
 }
+
+// baseFor reads the route of this row's attempt rather than the client's
+// initial service, which can differ after an authorized billing-door switch.
+func (facts recordFacts) baseFor(fallback string) string {
+	if facts.base != "" {
+		return facts.base
+	}
+	if facts.knobs.trace != nil && facts.knobs.trace.attemptBase != "" {
+		return facts.knobs.trace.attemptBase
+	}
+	return fallback
+}

internal/provider/client.go

@@ -1116,7 +1116,7 @@ func (c *Client) completionInOnePiece(
 			began: logBegan, status: status, served: served, err: err,
 			response: &response, reasoningTokens: reasoningTokens, responseBody: payload,
 		})
-		c.settle(ctx, c.modelFor(request), &response, receiptRefusalReason, len(responseText(&response)))
+		c.settle(ctx, c.modelFor(request), &response, receiptRefusalReason, len(responseText(&response)), knobs.trace)
 		return nil, false, err
 	}
 	// AND THE SAME SPLIT THE STREAM TAKES, taken over the whole body (answer.go).
@@ -1818,7 +1818,7 @@ func (c *Client) completeWithMessagesStreaming(
 		// An endpoint producing soup has failed this lineage as surely as one
 		// that went quiet, so the pin moves too.
 		c.releaseEndpoint(ctx, c.modelFor(request))
-		c.settle(ctx, c.modelFor(request), response, cut.Reason.word(), content.Len())
+		c.settle(ctx, c.modelFor(request), response, cut.Reason.word(), content.Len(), knobs.trace)
 		return nil, false, cut
 	}
 	for {
@@ -1866,7 +1866,7 @@ func (c *Client) completeWithMessagesStreaming(
 					response: response, reasoningTokens: reasoningTokens,
 					ttft: firstTokenAfter(began, firstToken),
 				})
-				c.settle(ctx, c.modelFor(request), response, cut.Reason.word(), content.Len())
+				c.settle(ctx, c.modelFor(request), response, cut.Reason.word(), content.Len(), knobs.trace)
 				return nil, false, cut
 			}
 			// A stream that broke off for any reason but a cut — the caller
@@ -1879,7 +1879,7 @@ func (c *Client) completeWithMessagesStreaming(
 				response: response, reasoningTokens: reasoningTokens,
 				ttft: firstTokenAfter(began, firstToken),
 			})
-			c.settle(ctx, c.modelFor(request), response, receiptTornReason, content.Len())
+			c.settle(ctx, c.modelFor(request), response, receiptTornReason, content.Len(), knobs.trace)
 			return nil, false, decodeErr
 		}
 		if response.ID == "" {
@@ -1933,7 +1933,7 @@ func (c *Client) completeWithMessagesStreaming(
 				ttft: firstTokenAfter(began, firstToken),
 			})
 			c.releaseEndpoint(ctx, c.modelFor(request))
-			c.settle(ctx, c.modelFor(request), response, receiptRefusalReason, content.Len())
+			c.settle(ctx, c.modelFor(request), response, receiptRefusalReason, content.Len(), knobs.trace)
 			return nil, false, refusal
 		}
 		if chunk.Usage != nil {
@@ -2147,7 +2147,7 @@ func (c *Client) completeWithMessagesStreaming(
 			response: response, reasoningTokens: reasoningTokens,
 			ttft: firstTokenAfter(began, firstToken),
 		})
-		c.settle(ctx, c.modelFor(request), response, cut.Reason.word(), content.Len())
+		c.settle(ctx, c.modelFor(request), response, cut.Reason.word(), content.Len(), knobs.trace)
 		return nil, false, cut
 	}
 	response.Choices = []ai.Choice{{Index: 0, FinishReason: finishReason, Message: ai.Message{
@@ -2164,7 +2164,7 @@ func (c *Client) completeWithMessagesStreaming(
 			response: response, reasoningTokens: reasoningTokens,
 			ttft: firstTokenAfter(began, firstToken),
 		})
-		c.settle(ctx, c.modelFor(request), response, receiptRefusalReason, content.Len())
+		c.settle(ctx, c.modelFor(request), response, receiptRefusalReason, content.Len(), knobs.trace)
 		return nil, false, err
 	}
 	// The last call has no successor to close it, so the clean end of the stream

internal/provider/dispatch.go

@@ -467,6 +467,9 @@ func (c *Client) send(ctx context.Context, request *ai.Request, knobs callKnobs,
 		// the end of it (calllog.go). The count lives on the knobs rather than
 		// here because a call that is repaired or relaxed comes back through
 		// this loop with a new body and the same trace.
+		if knobs.trace != nil {
+			knobs.trace.attemptBase = currentBase
+		}
 		knobs.trace.begin()
 		if attempt > 0 && switchedDoor {
 			// A BILLING-DOOR SWITCH IS NEITHER A RECONNECTION NOR A RETRY. The

internal/provider/funnel_law_test.go

@@ -345,12 +345,12 @@ func TestNothingOutsideTheFunnelTalksToAModelEndpoint(t *testing.T) {
 //	             which are request/response and carry no stream at all
 //	Fetch        sheetFetcher's GET of the lane sheet, which is not a model call
 //	             — it is the belief the choice is made from (lanes.go)
-//	fetchReceipt the bounded background GET for a cut stream's exact generation
+//	fetchReceiptAt the bounded background GET for a cut stream's exact generation
 //	             receipt; it creates no model work and never runs on the turn
 //
 // probeConnection is a credential-free HEAD of the configured origin. Its
 // contract tests forbid a prompt, a request body or redirect following.
-var funnelWireSenders = []string{"Fetch", "doEndpoint", "fetchReceipt", "probeConnection", "probeLane", "send"}
+var funnelWireSenders = []string{"Fetch", "doEndpoint", "fetchReceiptAt", "probeConnection", "probeLane", "send"}
 
 // funnelSendCallers is every function that reaches [Client.send].
 //

internal/provider/receipt.go

@@ -12,6 +12,8 @@ import (
 
 	"github.com/Agent-Field/agentfield/sdk/go/ai"
 	"github.com/Agent-Field/codeaf/internal/guard"
+	"github.com/Agent-Field/codeaf/internal/modelsource"
+	"github.com/Agent-Field/codeaf/internal/telemetry"
 )
 
 // THE RECEIPT: MONEY A CUT STREAM LEFT OFF THE WIRE.
@@ -99,6 +101,8 @@ type receiptWork struct {
 	result Reconciled
 	sink   ReconcileSink
 	queued time.Time
+	usage  func(int, int)
+	base   string
 }
 
 // deadline is the latest a receipt may be answered: [receiptFetchTimeout] after
@@ -157,7 +161,11 @@ func (m *receiptRouteMemo) heard(base string, now time.Time) {
 // measured first-frame in-band refusal is an upstream breaking before it
 // produced a generation, not money the provider charged. Those mutually
 // exclusive doors keep one call from ever being counted twice.
-func (c *Client) settle(ctx context.Context, model string, response *ai.Response, reason string, answerBytes int) {
+func (c *Client) settle(ctx context.Context, model string, response *ai.Response, reason string, answerBytes int, traces ...*callTrace) {
+	base := c.config.BaseURL
+	if len(traces) > 0 && traces[0] != nil && traces[0].attemptBase != "" {
+		base = traces[0].attemptBase
+	}
 	if response != nil && response.Usage != nil {
 		c.bill(ctx, model, response)
 		return
@@ -166,7 +174,7 @@ func (c *Client) settle(ctx context.Context, model string, response *ai.Response
 	// usage block supplies no measured figures to record, and calling the
 	// router's route on the vendor would send that vendor a request and bearer
 	// for a fact this client already knows cannot be there.
-	if c.config.Direct {
+	if c.config.Direct && (base == c.config.BaseURL || telemetry.RoutingProvider(base) != modelsource.DefaultID) {
 		return
 	}
 	sink := reconcileFrom(ctx)
@@ -193,7 +201,11 @@ func (c *Client) settle(ctx context.Context, model string, response *ai.Response
 		}
 		return
 	}
-	work := receiptWork{result: result, sink: sink}
+	work := receiptWork{result: result, sink: sink, base: base, usage: telemetry.CaptureUsageRecorder(telemetry.UsageDimensions{
+		RoutingProvider: telemetry.RoutingProvider(base),
+		ModelFamily:     telemetry.ModelFamily(result.Model), UsageStatus: "reported",
+		AccountingSource: "provider", ReceiptID: "generation:" + result.Ref,
+	})}
 	// THE WORK IS TOLD A RECEIPT IS OWED BEFORE IT IS QUEUED, and told it was
 	// answered only after the sink has banked it, so a caller waiting for its
 	// receipts cannot see zero owed while money is between the two
@@ -264,15 +276,18 @@ func (c *Client) nextReceipt() (receiptWork, bool) {
 // counted from when the receipt was queued ([receiptWork.deadline]).
 func (c *Client) reconcile(work receiptWork) {
 	result := work.result
-	base := strings.TrimRight(strings.TrimSpace(c.config.BaseURL), "/")
+	base := strings.TrimRight(strings.TrimSpace(work.base), "/")
+	if base == "" {
+		base = strings.TrimRight(strings.TrimSpace(c.config.BaseURL), "/")
+	}
 	if !receiptRoutes.askable(base, time.Now()) {
 		work.sink(result)
 		return
 	}
 	ctx, cancel := context.WithDeadline(context.Background(), work.deadline())
 	defer cancel()
 	for attempt := 0; attempt < receiptAttempts; attempt++ {
-		billed, found, noRoute := c.fetchReceipt(ctx, result.Ref)
+		billed, found, noRoute := c.fetchReceiptAt(ctx, result.Ref, base)
 		if noRoute {
 			receiptRoutes.heard(base, time.Now())
 			work.sink(result)
@@ -283,6 +298,12 @@ func (c *Client) reconcile(work receiptWork) {
 			billed.Model = result.Model
 			result.Billed = billed
 			result.Found = true
+			// Recovery is reached only when the stream had no usage block, so
+			// these tokens replace an unknown receipt rather than add to a
+			// previously reported total.
+			if work.usage != nil {
+				work.usage(billed.PromptTokens, billed.CompletionTokens)
+			}
 			work.sink(result)
 			return
 		}
@@ -312,7 +333,12 @@ type receiptWire struct {
 // receipt from the one durable capability answer; every other failure simply
 // leaves both false so the caller may follow its bounded retry schedule.
 func (c *Client) fetchReceipt(ctx context.Context, ref string) (Billed, bool, bool) {
-	endpoint, err := url.Parse(strings.TrimSpace(c.config.BaseURL))
+	return c.fetchReceiptAt(ctx, ref, c.config.BaseURL)
+}
+
+// fetchReceiptAt follows the route the generation actually used.
+func (c *Client) fetchReceiptAt(ctx context.Context, ref, base string) (Billed, bool, bool) {
+	endpoint, err := url.Parse(strings.TrimSpace(base))
 	if err != nil {
 		return Billed{}, false, false
 	}

internal/provider/receipt_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/http"
 	"net/http/httptest"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -939,3 +940,35 @@ func TestAReceiptIsAnsweredWithinReceiptWaitOfBeingQueued(t *testing.T) {
 		t.Fatalf("the receipt was answered %v after it was queued, past ReceiptWait %v", answered, ReceiptWait)
 	}
 }
+
+// The actual receipt wire door retains both response and caller-context bounds
+// when recovery selects the route of an overflow attempt.
+func TestActualReceiptWireDoorKeepsResponseAndDeadlineBounds(t *testing.T) {
+	t.Run("response size", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = fmt.Fprint(w, strings.Repeat(" ", maxReceiptBytes)+`{"data":{"tokens_prompt":10,"tokens_completion":3}}`)
+		}))
+		t.Cleanup(server.Close)
+		client := receiptTestClient(t, server)
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		if _, found, _ := client.fetchReceiptAt(ctx, "oversized", server.URL); found {
+			t.Fatal("receipt read beyond maxReceiptBytes")
+		}
+	})
+	t.Run("caller deadline", func(t *testing.T) {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() }))
+		t.Cleanup(server.Close)
+		client := receiptTestClient(t, server)
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+		began := time.Now()
+		if _, found, _ := client.fetchReceiptAt(ctx, "waiting", server.URL); found {
+			t.Fatal("expired receipt was reported found")
+		}
+		if time.Since(began) > time.Second {
+			t.Fatal("receipt wire door ignored caller deadline")
+		}
+	})
+}