let's work on https://linear.app/entirehq/issue/COR-985/phase-3-converge-remaining-nats-publishing-and-redelivery-code-onto-go

mirror-pipeline: NATS/otel migration survey (COR-985)

Important upfront discrepancy

The ticket targets go-nuts v0.8.0 (packages natsmsg, backoff, jsconsumer), but the repo currently pins v0.7.0, and vendor/github.com/entireio/go-nuts/ contains only the root nuts package (connect.go, drain.go, shutdown.go, doc.go). The three subpackages named in the ticket are not present — they only appear as doc links in vendor/.../doc.go:25-27. So step one of the migration is a version bump + re-vendor before any of natsmsg/backoff/jsconsumer can be imported.

1. go-nuts version + where it's imported

• Pinned version: github.com/entireio/go-nuts v0.7.0 — go.mod:9 (also go.sum:28-29).
• Module import path is the root package github.com/entireio/go-nuts (package name nuts). Only these symbols are used today: Connect, Drain, NewShutdownGroup, IsShutdownFetchErr, and options WithName/WithNATSOptions/WithRetryOnFailedConnect.
• Import + usage sites (all nuts.*):
  • cmd/worker/main.go:16,236,245-247 (NewShutdownGroup, Connect, WithName, WithNATSOptions); cmd/worker/consumer.go:11,167 (IsShutdownFetchErr)
  • cmd/fanout/main.go:25,126,131 (Connect, Drain)
  • cmd/core-fanout/main.go:23,105,110
  • cmd/meta-fanout/main.go:27,172,177; cmd/meta-fanout/backfill.go:20,141,146; cmd/meta-fanout/backfill_prs.go:16,131,136
  • cmd/trails-fanout/main.go:25,93,98
  • cmd/webhook-ingest/main.go:21,55,60
  • cmd/webhook-forwarder/main.go:23,76,81; cmd/webhook-forwarder/forwarder.go:16,184 (IsShutdownFetchErr)
  • cmd/rekicker/main.go:27,105
  • cmd/loadgen/main.go:28,208; cmd/loadtest-jetstream/main.go:26,87
  • cmd/mirror-pipeline-admin/nats.go:14,37-39
  • internal/natsregions/natsregions.go:27,197-199 (Connect, WithRetryOnFailedConnect, WithNATSOptions)
  • pkg/fanoutengine/engine.go:27,355,410 (IsShutdownFetchErr)

2. pkg/otelnats — full inventory (pkg/otelnats/nats.go, 57 lines)

Purpose: W3C trace-context propagation over nats.Header (no span creation of its own). This is the exact functional overlap with go-nuts/natsmsg's header propagation.

Exported symbols:

• type HeaderCarrier nats.Header (:15) — adapts nats.Header to propagation.TextMapCarrier. Methods Get (:17, returns first value), Set (:25), Keys (:29). Compile-time assertion var _ propagation.TextMapCarrier at :56.
• func Inject(ctx, msg *nats.Msg) (:39) — creates msg.Header if nil, then otel.GetTextMapPropagator().Inject into the carrier. Writes traceparent/baggage.
• func Extract(ctx, msg *nats.Msg) context.Context (:48) — no-op if msg.Header == nil; otherwise otel.GetTextMapPropagator().Extract.

No attributes, no span naming — pure propagation. Uses the global propagator (otel.GetTextMapPropagator()).

Callers (file:line):

• internal/natspub/natspub.go:22 (import), :106 Inject (inside Publish)
• pkg/webhookdlq/publisher.go:16, :112 Inject
• cmd/meta-fanout/lifecycle.go:22, :152 Inject (producer, publishOne), :264 Extract (consumer handleMessage)
• cmd/worker/consumer.go:20, :187 Extract
• cmd/webhook-forwarder/forwarder.go:23, :197 Extract
• pkg/fanoutengine/engine.go:34, :430 Extract (handleMessage)
• Test: pkg/otelnats/nats_test.go (whole file, package otelnats)

So otelnats has 7 production call sites across 6 files plus 1 test file. Every consumer's Extract and every producer's Inject funnels through it.

3. internal/natspub — full inventory (internal/natspub/natspub.go, 131 lines)

Purpose: the shared JetStream publish core behind all typed job/event publishers — one producer span, dedup header, bounded pub-ack wait, one INFO log. This is what the ticket wants routed through natsmsg's "shared legacy publishing adapter."

Exported API:

• const MsgIDHeader = "Nats-Msg-Id" (:26) — JetStream dedup header.
• const AckWait = 5 * time.Second (:35) — bounds the synchronous pub-ack wait (see doc :27-34: converts a reconnect stall into a caller error → Nak → redelivery).
• type JetStreamPublisher interface { PublishMsg(*nats.Msg, ...nats.PubOpt) (*nats.PubAck, error) } (:38) — the publish subset of nats.JetStreamContext.
• type Publication struct (:44) — fields: Tracer, SpanName, ErrKind, LogMsg, Subject, MsgID, Marshal func() ([]byte, error), SpanAttrs []attribute.KeyValue, LogAttrs []slog.Attr.
• func PlacementAttrs(jurisdiction, cluster string) []attribute.KeyValue (:72) — standard entire.target_jurisdiction / entire.target_cluster_id attrs.
• func Publish(ctx, js JetStreamPublisher, pub Publication) error (:80) — the core.

What Publish does (span naming, msg-id, PubAck, timeouts):

• Starts a producer span pub.SpanName from otel.Tracer(pub.Tracer) with SpanKindProducer (:87-90). Standard attrs messaging.system=nats, messaging.operation.type=publish, messaging.destination.name=<subject> then domain SpanAttrs.
• Runs pub.Marshal() inside the span; on error records span.RecordError + SetStatus(Error, "marshal <ErrKind>") and returns fmt.Errorf("marshal %s: %w", ...) (:93-98).
• Builds *nats.Msg, sets Nats-Msg-Id header = pub.MsgID, calls otelnats.Inject(ctx, msg) (:100-106), sets messaging.message.id attr.
• Bounded pub-ack: context.WithTimeout(ctx, AckWait) then js.PublishMsg(msg, nats.Context(pubCtx)) (:109-111).
• On publish error: RecordError + SetStatus(Error, "nats publish"), returns fmt.Errorf("publish %s: %w", ...) (:112-116).
• On success: sets messaging.nats.sequence (from ack.Sequence) and messaging.nats.duplicate (from ack.Duplicate) span attrs (:118-121), and logs one INFO with LogAttrs + subject/msg_id/sequence/duplicate (:123-129).

Legacy vs modern JetStream API: legacy nats.JetStreamContext.PublishMsg (the github.com/nats-io/nats.go v1 API), not the modern jetstream package.

Callers (file:line) — two categories:

(a) Full delegation via natspub.Publish (typed publishers; each aliases NatsMsgIDHeader = natspub.MsgIDHeader and type JetStreamPublisher = natspub.JetStreamPublisher):

• internal/mirrorjobs/publisher.go:7,14,17,40
• pkg/prjobs/publisher.go:6,12,15,31,39
• pkg/donejobs/publisher.go:6,12,15,31,39
• pkg/pushauthorjobs/publisher.go:9,15,18,34,42
• pkg/trailforgeevents/publisher.go:6,11,14,21,29
• pkg/metajobs/publisher.go:6,12,15,31,39
• pkg/webhookevents/publisher.go:13,22,89 (wraps Publish with an extra latency metric; see :66 description referencing natspub.AckWait)

(b) Uses only natspub.AckWait / the type alias but hand-rolls its own publish (these intentionally diverge from Publish — see notes below):

• pkg/webhookdlq/publisher.go:15,31,35 — publishAckWait = natspub.AckWait; hand-rolls publish at :84-146 because capture detaches from caller ctx (context.Background(), :125) and needs the PubAck for the advisory backstop's dedup check (doc :26-31).
• cmd/meta-fanout/lifecycle.go:20,46,51 — lifecyclePublishAckWait = natspub.AckWait; hand-rolls publishOne at :128-167 (per-region fan, header carried verbatim + re-inject).

Test: internal/natspub/natspub_test.go (fake fakeJetStream implementing PublishMsg, tracetest.SpanRecorder).

4. Custom worker/fanout fetch loops (keep-local candidates)

There are two families: a shared engine and per-binary hand-rolled loops. All use the legacy PullSubscribe + sub.Fetch(1|batch, nats.MaxWait(...)) API and depend on nuts.IsShutdownFetchErr + otelnats.Extract.

A. pkg/fanoutengine/engine.go — the extracted shared runtime (doc :1-14 explicitly says it is deliberately kept on the "legacy nats.JetStreamContext API and OTel wiring… The modern jetstream.Consume shape lands when this graduates to the standalone module"). Key pieces the migration must preserve:

• Subscribe+backoff loop Run (:290-330), ensureInactiveThreshold (:255-282, UpdateConsumer reconcile).
• Two fetch loops: serial consumeLoop Fetch(1) (:339-365) and worker-pool consumeLoopPool (:376-423) — unbuffered work channel, FetchBatch+Concurrency bound, drains channel to close on shutdown.
• Hooks struct (:111-123): RecordDispatch/RecordPanic/IncActive/DecActive/RecordFetchBatch/RecordDuration — the batching + metrics callbacks jsconsumer would need to express.
• Ack/Nak/Term decision tree Dispatch/dispatchRoutes/nakOrExhaust/handlePanic (:457-710), flat NakWithDelay(cfg.NakDelay), TermOnExhaustion (COR-762), DLQ capture, route memo.
• Uses otelnats.Extract (:430) and nuts.IsShutdownFetchErr (:355,:410).
• Consumers built on it: cmd/fanout (worker pool + DLQ), cmd/core-fanout, cmd/meta-fanout (live webhook dispatcher + github_meta backfill via TermOnExhaustion), cmd/meta-fanout backfill-prs (backfill_prs_consumer.go), cmd/trails-fanout. See cmd/*/consumer.go (heads confirm fanoutengine.New(...)).

B. cmd/worker/consumer.go — hand-rolled, NOT on the engine. runSubscription (:117-155) + consumeLoop Fetch(1) (:157-180). This has the richest semantics the ticket flags as keep-local:

• Heartbeat: startInProgressExtender (:718-740) periodically calls msg.InProgress() on consumerInProgressInterval = consumerAckWait/3 (:33) to extend the ack deadline during long git syncs; idempotent stop.
• consumerAckWait = 15m, consumerMaxDeliver = 10, consumerBackoffInit/Max, defaultSyncTimeout = 30m.
• Disposition tree disposeResult with dispositionAck/Noop/Term/Nak/Raced (:391-485), publish-before-ack (publishReadyOrRetry), failure-index integration.
• Uses otelnats.Extract (:187), nuts.IsShutdownFetchErr (:167), span mirror.consume.

C. cmd/meta-fanout/lifecycle.go — hand-rolled lifecycle consumer (lifecycleConsumer.run :207-242, consumeLoop Fetch(1) :244-261). Note doc at cmd/meta-fanout/consumer.go:14-21 calls this and the pr-backfill "still-hand-rolled." Does not call nuts.IsShutdownFetchErr (just logs+returns on fetch error, :250-255). Uses otelnats.Extract (:264) and hand-rolled producer publish publishOne (:128-167) with otelnats.Inject (:152).

D. cmd/webhook-forwarder/forwarder.go — hand-rolled, NOT on engine (mirrors it). run (:131-172) + consumeLoop Fetch(1) (:174-194), ensureInactiveThreshold (:108-129). Uses otelnats.Extract (:197), nuts.IsShutdownFetchErr (:184), DLQ via webhookdlq.Publisher.

Batching/hooks/heartbeat semantics that jsconsumer would need to match (per ticket "default keep local"): the worker's InProgress heartbeat cadence tied to a 15m AckWait + 30m sync timeout; fanoutengine's unbuffered worker pool with FetchBatch<=Concurrency AckWait-correctness bound + 6 metric hook callbacks + TermOnExhaustion + route-memo. These are the "added interface complexity" the ticket warns about.

5. Tests + how they run

• No embedded NATS server anywhere. No nats-server/natsserver/server.Options imports outside vendor; no nats.Connect/JetStream() in any _test.go. Tests drive Dispatch/dispose logic through narrow interface seams (ackableMsg/AckableMsg/inProgressMsg/messageFetcher/JetStreamPublisher) with in-memory fakes.
• otelnats tests: pkg/otelnats/nats_test.go — HeaderCarrier get/set/keys, Inject/Extract round-trip via a real sdktrace.NewTracerProvider + W3C composite propagator, nil-header no-op cases.
• natspub tests: internal/natspub/natspub_test.go — fakeJetStream PublishMsg fake + tracetest.SpanRecorder; asserts subject/msg-id/payload/traceparent, producer span name/kind/attrs (messaging.*, entire.target_*, sequence, duplicate), marshal-error and nats-error span status.
• Consumer tests (fakes): cmd/worker/consumer_*_test.go (failindex, panic, span, completion, etc.), pkg/fanoutengine/{engine,pool,resolver,routememo}_test.go, cmd/{fanout,core-fanout,trails-fanout,meta-fanout}/consumer_test.go + dlq_test.go + *_panic_test.go, cmd/webhook-forwarder/forwarder*_test.go, cmd/meta-fanout/{backfill_consumer,backfill_prs_consumer,lifecycle,mirror_seed}_test.go.
• How tests run: no Makefile. Tooling is mise (mise.toml):
  • mise run test → go test ./... (mise.toml:34-36)
  • mise run test:race → go test -race ./... (:38-40)
  • mise run build → go build ./cmd/...; mise run vet → go vet ./...; mise run lint (file tasks under mise-tasks/lint/{go,gofmt}).
  • CI: .github/workflows/test.yml runs mise run build, vet, test:race, gofmt check, lint on every PR (via jdx/mise-action@v2).

6. Other hand-rolled NATS tracing/publishing outside otelnats/natspub

Hand-rolled producer spans that DO use otelnats.Inject (must stay behavior-equivalent if natsmsg replaces otelnats):

• pkg/webhookdlq/publisher.go:84-146 — span webhook.dlq.publish (SpanKindProducer, :96), full messaging.* + entire.dlq.*/github.* attrs, Nats-Msg-Id (:111), otelnats.Inject (:112), PubAck sequence/duplicate; deliberately publishes on context.Background() (:125) and returns the PubAck for the advisory backstop. Advisory backstop lives in pkg/webhookdlq/advisory.go.
• cmd/meta-fanout/lifecycle.go:128-167 — span resource_lifecycle.fan (SpanKindProducer, :130), otelnats.Inject (:152), header carried verbatim + Entire-Lifecycle-Fanned-By echo-guard.

Hand-rolled consumer spans (create messaging.system=nats receive spans directly, all via otelnats.Extract): cmd/worker/consumer.go:189 (mirror.consume), cmd/webhook-forwarder/forwarder.go:198 (webhook.forward.dispatch), cmd/meta-fanout/lifecycle.go:265 (meta_fanout.lifecycle_fan), pkg/fanoutengine/engine.go:431 (per-config <tracer>.dispatch).

Raw PublishMsg that sets Nats-Msg-Id but does NO tracing / NO trace-context injection (candidates the migration might standardize, or explicitly leave alone — mostly CLI/admin/backfill):

• cmd/meta-fanout/backfill.go:88-101 (backfillPublisher.Enqueue) — sets Nats-Msg-Id, bounded ctx, no span/Inject.
• cmd/meta-fanout/backfill_prs.go:75-90 (prBackfillPublisher.Enqueue) — same.
• cmd/mirror-pipeline-admin/dlq.go:570-582 (publishReplay) — replay, sets Nats-Msg-Id, no tracing.
• cmd/mirror-pipeline-admin/rekick.go:109 and cmd/rekicker/rekick.go:223 — publish failindex.RekickMsg (pkg/failindex/failindex.go:47-51), which sets Mirror-Rekick + a fresh Nats-Msg-Id via nuid.Next(); no trace context.
• cmd/loadgen/publisher.go, cmd/loadtest-jetstream/main.go — load tools, no otelnats.

The HeaderCarrier in otelnats (nats.Header) is distinct from the vendored propagation.HeaderCarrier (http.Header) — no other bespoke TextMapCarrier exists in the repo; otelnats.HeaderCarrier is the single NATS-header carrier.

Migration-impact summary

• natsmsg replaces pkg/otelnats (propagation) and the natspub.Publish core — but note 2 hand-rolled publishers (webhookdlq, lifecycle.publishOne) only borrow natspub.AckWait/otelnats.Inject and have semantics (context.Background detach + PubAck for backstop; verbatim header carry + echo-guard) the ticket's "legacy adapter" must still support or these stay local.
• backoff (NakWithDelay policy, Term-on-final-delivery) maps onto the flat-NakWithDelay + TermOnExhaustion logic in pkg/fanoutengine/engine.go and the 6 hand-rolled NakWithDelay sites (worker, forwarder, lifecycle, fanout/core-fanout/trails-fanout/meta-fanout consumers).
• jsconsumer would target the fetch loops in §4 — but the worker's InProgress heartbeat (consumerAckWait/3) and fanoutengine's worker-pool batching + hooks are exactly the "batching/hooks/heartbeat" the ticket says to keep local by default.
• Prerequisite: bump go.mod from v0.7.0 → v0.8.0 and re-vendor (the 3 subpackages are not vendored yet).</result> <usage><subagent_tokens>133080</subagent_tokens><tool_uses>28</tool_uses><duration_ms>232226</duration_ms></usage>

COR-985 go-nuts migration: reconnaissance of entire-api

Repo root: /Users/nodo/work/tasks/COR-985-go-nuts/entire-api (module github.com/entirehq/entire-api, Go 1.26.4). On branch main at 3d1ff48f.

1. go-nuts version and existing import surface

Pinned version: v0.7.0 — NOT the v0.8.0 the ticket targets.

• go.mod:29: github.com/entireio/go-nuts v0.7.0
• go.sum:109-110: only v0.7.0 hashes present. (Module cache has up to v0.7.0; v0.8.0 is not yet downloaded, so the bump will need go get.)

go-nuts is already imported widely. Three sub-packages are in use:

• natsmsg — githubpr, githubmeta, userlifecycle, forwarder (consumer + publisher), plus analysis/jobs.go, ingest/pipeline.go, ingest/jobs.go, jobsdlq/jobsdlq.go, reattribute/consumer.go and DLQ test files.
• jsconsumer — settingsconsumer, mirrorlifecycle, repolifecycle, repogroupprimary, trailforgeevents. (These five are already on the modern shared-consumer pattern — they are the template for the migration, but are out of scope.)
• root package github.com/entireio/go-nuts (aliased nuts, e.g. nuts.Drain) — server/server.go, server/enqueue_analysis.go, server/analysis_replay.go, server/backfill.go.
• backoff — NOT imported anywhere yet in entire-api.

The go-nuts v0.7.0 API the migration will consume (from the module cache /Users/nodo/go/pkg/mod/github.com/entireio/go-nuts@v0.7.0):

• jsconsumer.Config (fields: Stream, Durable, FilterSubject/FilterSubjects, AckWait, MaxDeliver, SpanName, Name, Tracer, InactiveThreshold, MaxAckPending, KeepInProgress, MaxMessages), jsconsumer.Start→*Runner, jsconsumer.Run, and generic jsconsumer.Process[E any](ctx, msg, cfg, decode, onEvent, onUndecodable).
• backoff.Policy{NakDelay, Factor, MaxDelay, MaxDeliver, TermOnExhaustion} with NakOrTerm(msg), DelayFor(n), NumDelivered, IsFinalDelivery. Delay formula (backoff/backoff.go:101): flat NakDelay when Factor<=1 or n<=1; else NakDelay × Factor^(n-1) capped at MaxDelay.
• natsmsg: StartConsumerSpan, Inject, Extract, DeadLetter, SubjectToken, DLQ header consts, Publisher, ClampToInt64.

2. The four in-scope consumers/publishers

All four currently use the legacy hand-rolled pattern (raw jetstream.New(nc) + CreateOrUpdateConsumer + cons.Consume + a hand-written handle), NOT jsconsumer. All use the modern jetstream.JetStream API (via github.com/nats-io/nats.go/jetstream) — none use the legacy nats.JetStreamContext. All share identical config knobs: AckPolicy: AckExplicitPolicy, AckWait: 30s, MaxDeliver: 8 (const consumerMaxDeliver/activityMaxDeliver = 8). None sets MaxAckPending or InactiveThreshold. None uses NakWithDelay, backoff.Policy, or any DLQ — poison messages are Term()ed with no capture; transient failures are a plain msg.Nak() (redelivery paced by the 30s AckWait, bounded by MaxDeliver=8, then JetStream silently drops).

githubpr — CONSUMER only

• Files: internal/githubpr/consumer.go (166), contract.go (98, stream/subject/Event), metrics.go (31), consumer_test.go (137).
• Durable github-pr-consumer; Stream github_pr_v1; FilterSubject github_pr.v1.<jur>.<cluster> (SubjectFor, contract.go:39). AckWait 30s, MaxDeliver 8 (consumer.go:72-78).
• Consume loop consumer.go:82 cons.Consume(func(msg){ c.handle(ctx,msg) }).
• Disposition (handle, lines 105-143): undecodable JSON → Term() (line 117); missing RepoULID/Number → Term() (125); store upsert error → Nak() (135); success → Ack() (139).
• Tracing: natsmsg.StartConsumerSpan(ctx, otelsetup.Tracer(), msg, "github_pr.consume") (line 107); baggage via otelsetup.WithBaggage(ctx, otelsetup.RepoIDAttr(...)) (128).

githubmeta — CONSUMER only

• Files: internal/githubmeta/consumer.go (169), contract.go (99), metrics.go (31), consumer_test.go (175).
• Durable github-meta-consumer; Stream github_meta_v1; FilterSubject github_meta.v1.<jur>.<cluster>. AckWait 30s, MaxDeliver 8 (consumer.go:72-78).
• Disposition (handle, 105-140): undecodable → Term() (117); missing RepoULID → Term() (124); upsert error → Nak() (133); success → Ack() (137). Nearly byte-identical to githubpr.
• Tracing: natsmsg.StartConsumerSpan(...,"github_meta.consume") (107); otelsetup.WithBaggage/RepoIDAttr (127).

userlifecycle — CONSUMER only

• Files: internal/userlifecycle/consumer.go (145), contract.go (52), metrics.go (33), consumer_test.go (116).
• Durable user-lifecycle-consumer; Stream resource_lifecycle_v1; uses FilterSubjects: []string{subjectUser} where subjectUser = "resource_lifecycle_v1.user.*" (contract.go:35) — note the plural FilterSubjects slice, unlike the other three's singular FilterSubject. AckWait 30s, MaxDeliver 8 (consumer.go:65-71).
• Disposition (handle, 97-145): undecodable → Term() (109); ResourceType != "user" → Ack()+skip metric (116, defensive); missing ResourceID → Term() (123); on op == "deleted" store DeleteUserData error → Nak() (133), success → Ack() (137); any other op → Ack()+skip (143).
• Tracing: natsmsg.StartConsumerSpan(...,"userlifecycle.consume") (99); no baggage helper here.

forwarder — CONSUMER and PUBLISHER (the most complex; contains the only real delay/retry envelope logic)

• Files: internal/forwarder/consumer.go (230), publisher.go (367), subjects.go (42, stream/subject), metrics.go (37), plus tests: forwarder_test.go (156, residency assertion), publisher_test.go (282, fakes), integration_test.go (202, real NATS).
• Consumer (consumer.go): durable activity-consumer; Stream user_activity_v1; FilterSubject user_activity_v1.<jur> (SubjectUserActivityFor). AckWait 30s, MaxDeliver 8 (consumer.go:83-89). Has a SingleRegion bool residency flag.
  • Disposition (handle, 116-220): undecodable → Term() (127); missing Recipient → Term() (132); !recipientIsLocal → Ack()+drop (144, fail-closed residency); facet-enrichment branch (Type==checkpoint_facets) → EnrichMyActivityFacets, error Nak() (157)/success Ack() (161); normal upsert UpsertMyActivity error Nak() (215)/success Ack() (219).
  • Hand-rolled seq guard: reads msg.Metadata() for md.Sequence.Stream → row.StateSeq (consumer.go:207-209).
  • Tracing: natsmsg.StartConsumerSpan(...,"user_activity.consume") (118); otelsetup.WithBaggage with Recipient/RepoID/ActivityType attrs (137-138).
• Publisher (publisher.go): interface JSPublisher = PublishMsg(ctx, *nats.Msg, ...jetstream.PublishOpt) (*jetstream.PubAck, error) (158-160); holds conns map[string]JSPublisher keyed by jurisdiction. PublishUserActivity (199-316) returns nil=handled/permanent-skip (caller Acks) vs err=transient (caller Naks). Bounds each publish with its own publishTimeout = 5s context (publisher.go:31, 306).
  • Hand-rolled message-ID / dedup: dedupeID(ev) (publisher.go:333-353) builds the Nats-Msg-Id header string (type:recipient:origin:sha[:cpID][:state][:facetKind]); set at publisher.go:284. facetDedupeKind (358-367).
  • Hand-rolled producer span (NOT via natsmsg): otelsetup.Tracer().Start(ctx, "user_activity.publish", trace.WithSpanKind(SpanKindProducer), ...) (publisher.go:203), then natsmsg.Inject(ctx, msg) (292) to propagate trace context into headers.
  • Note: this is a publish-path retry envelope expressed as a return-value contract to the source consumer (ingest/analysis Nak the upstream message), not a JetStream NakWithDelay. There is no numeric backoff/delay formula in the four — the only publish delay is the flat 5s publishTimeout.

3. Shared internal helpers used by the four (blast radius)

Two internal helper surfaces, plus the go-nuts natsmsg package:

internal/otelsetup (attr.go, baggage.go, setup.go) — used by ALL FOUR and 13 other files. Helpers the four call: otelsetup.Tracer(), otelsetup.WithBaggage(), and attr constructors RepoIDAttr / RecipientAttr / HomeJurisdictionAttr / ActivityTypeAttr (attr.go:18-28). Blast radius of changing otelsetup is large — 17 non-test importers including ingest, analysis, reattribute, settingsconsumer, mirrorlifecycle, repolifecycle, repogroupprimary, trailforgeevents, server. The migration should keep passing otelsetup.Tracer() into jsconsumer.Config.Tracer (as settingsconsumer/consumer.go:59 already does) so trace attribution/span names are unchanged.

go-nuts natsmsg — StartConsumerSpan (all four consumers) and Inject (forwarder publisher). Also used non-trivially by analysis/jobs.go, ingest/pipeline.go+jobs.go (DeadLetter/DLQ), jobsdlq/jobsdlq.go (DLQ helpers), reattribute/consumer.go. This is the library being upgraded, so behavior change here hits ingest/analysis/reattribute/jobsdlq too — verify v0.8.0 keeps StartConsumerSpan/Inject/DeadLetter/DLQ header consts signature-compatible.

Per-package metrics.go — each of the four has its own local OTel counters (e.g. githubpr.written/.dropped, github.com/entirehq/entire-api/internal/githubpr meter). These are self-contained (not shared) and must be preserved through the migration since jsconsumer.Process's onUndecodable callback is where the dropped counter currently fires (compare settingsconsumer/consumer.go:115-117).

4. Tests and how they run

Per-package tests:

• githubpr/consumer_test.go, githubmeta/consumer_test.go, userlifecycle/consumer_test.go — pure unit tests with an in-memory fake jetstream.Msg (records acked/naked/termed) and a fake store. No NATS. githubpr/githubmeta fakes already implement NakWithDelay, TermWithReason, DoubleAck, InProgress (so they're jsconsumer-ready); userlifecycle's fake embeds jetstream.Msg and only overrides Ack/Nak/Term.
• forwarder/forwarder_test.go — residency assertion (TestUserActivityEventIsContentFree), reflection-only, no NATS.
• forwarder/publisher_test.go — fake JSPublisher (fakeConn), no NATS.
• forwarder/integration_test.go — TestIntegrationCrossClusterDelivery: requires two external Docker NATS-JS servers via env ACTIVITY_TEST_NATS_A/_B; self-skips when unset. Not embedded.

Embedded in-process NATS (nats-server/v2/server) is used only in internal/multicellular/harness/nats.go, internal/jobsdlq/sweeper_test.go, internal/ingest/jobs_dedupe_test.go — NOT in the four's own tests.

Cross-cell coverage (internal/multicellular): the harness starts REAL embedded NATS per cell and wires forwarder.NewConsumer (harness.go:164), githubmeta.NewConsumer (165), and forwarder.NewPublisher (755). githubpr and userlifecycle are NOT exercised in multicellular (no githubpr./userlifecycle. references there; github_pr_v1 isn't even in the harness cellStreams list). So for those two, the per-package unit tests are the only coverage.

How tests run (mise; scripts under mise-tasks/, : = dir separator):

• mise run test → go test ./... (unit only — covers all four's unit tests).
• mise run test:race → go test -race ./....
• mise run test:multicellular → ACTIVITY_EMBEDDED_PG=1 go test ./internal/multicellular/... -v (embedded PG, in-process NATS bus).
• mise run test:integration → store tests vs throwaway Docker Postgres.
• mise run test:pipeline → httpapi integration vs Docker Postgres + Docker NATS-JS (-js).
• No plain Makefile; CLAUDE.md mandates mise run build, mise run vet, mise run lint before pushing.

5. Out-of-scope packages (do NOT touch)

• reattribute — internal/reattribute/ (consumer.go, contract.go, metrics.go, consumer_test.go). Legacy consumer on resource_lifecycle_v1; uses natsmsg.StartConsumerSpan at consumer.go:177. (Ticket: heartbeat/long-handler follow-up.)
• analysis — internal/analysis/ (jobs.go, analyze.go, forward.go, metrics.go + tests). Uses natsmsg.Inject, natsmsg.StartConsumerSpan, and natsmsg.DeadLetter/DLQ (jobs.go:126,258,379). (Ticket: heartbeat/long-handler follow-up.)
• ingest — internal/ingest/ (large: pipeline.go, jobs.go, flow.go, checkpoint*.go, etc.). Uses natsmsg DeadLetter/DLQ and spans (pipeline.go:208, jobs.go:326,520,540). (Ticket: keep ingest batching local.)
• Related but also out of scope: internal/jobsdlq/ (DLQ sweeper built on natsmsg.DeadLetter).

Note the shared coupling caveat: since analysis, ingest, reattribute, and jobsdlq all depend on go-nuts natsmsg (and otelsetup), the v0.7.0→v0.8.0 bump touches their compile surface even though their code is out of scope — a compile/regression check across these packages is warranted after the version bump.</result> <usage><subagent_tokens>112756</subagent_tokens><tool_uses>42</tool_uses><duration_ms>290426</duration_ms></usage> </task-notification>

COR-985 go-nuts migration — repo exploration report

Repo: /Users/nodo/work/tasks/COR-985-go-nuts/entiredb (module entire.io, Go 1.26.4)

1. go-nuts version pinned & vendoring

• Pinned version: github.com/entireio/go-nuts v0.7.0 — go.mod:32; go.sum:264-265. The migration target is v0.8.0, so a go.mod/go.sum bump is required.
• No vendoring. There is no vendor/ directory at the repo root (confirmed by ls). Module-mode build against the module cache; the resolved package lives at /Users/nodo/go/pkg/mod/github.com/entireio/go-nuts@v0.7.0.
• Note: v0.8.0 is not in the local module cache (only v0.3.0 … v0.7.0 present), so the new "legacy publishing adapter" cannot be inspected locally — it must be a v0.8.0 addition. In v0.7.0, natsmsg.Publisher (natsmsg/publish.go:43-102) wraps only a modern jetstream.JetStream (p.JS.PublishMsg(pubCtx, msg)), so there is no legacy nats.JetStreamContext adapter in the pinned version.

2. Redelivery delay / final-delivery / disposition under repoops and permswebhook

2a. repoops — /Users/nodo/work/tasks/COR-985-go-nuts/entiredb/internal/entirecore/repoops/consumer.go

Constants (consumer.go:47-70): defaultAckWait = 2*time.Minute, defaultMaxDeliver = -1 (unlimited, all ops), defaultNakDelay = 5*time.Second, defaultFetchBatch = 1, nakDelayMax = 30*time.Second.

Delay formula — nakDelay(numDelivered uint64) (consumer.go:848-861):

• base = NakDelay (5s), exponent = geometric ×2 per delivery (NakDelay << (numDelivered-1) = NakDelay·2^(n-1)), cap = nakDelayMax (30s). deliveryCount (consumer.go:866-872) reads meta.NumDelivered, defaulting to 1 when metadata is unavailable.
• MaxDeliver handling: default -1 (unlimited); set into jetstream.ConsumerConfig.MaxDeliver (consumer.go:301-309) and jsconsumer.Config.MaxDeliver (consumer.go:317-327). No client-side "final delivery" detection — because delivery is unlimited, poison removal is done by Term→DLQ, not by exhaustion.

Disposition logic:

• handleMessage (consumer.go:370-454) → nak on executor error, terminate on structural error, Ack on success/skip.
• nak (consumer.go:826-840):

• DLQ + Term ordering — disposePoison (consumer.go:767-785), routed to by terminate (bad subject/body/op, consumer.go:790-803) and recoverPanic (consumer.go:810-824):

• DLQ publish ordering: publish-to-DLQ FIRST, then Ack. If the DLQ publish fails, it NakWithDelay(c.cfg.NakDelay) (flat base delay, not the geometric nakDelay) so poison is never dropped without a captured copy. DLQ subject = repo.ops.dlq.v1.<op>.<reason> (dlqSubjectPrefix = "repo.ops.dlq.v1.", consumer.go:45). Note the comment at consumer.go:761-766 explaining Term/MaxDeliver do not delete on a WorkQueue stream, hence the DLQ+Ack pattern (COR-944).
• The DLQ is natsmsg.DLQPublisher (consumer.go:178, required in New at consumer.go:252-254), wired in repoops_wiring.go:370-378 as DLQ: js where js is the legacy nats.JetStreamContext from nc.JetStream() (repoops_wiring.go:341). natsmsg.DLQPublisher is PublishMsg(m *nats.Msg, opts ...nats.PubOpt) (*nats.PubAck, error) (go-nuts natsmsg/deadletter.go:32-33).
• Disposition metrics: metrics.go — repoops.jobs_naked / jobs_terminated / jobs_skipped / jobs_completed, plus skip-reason constants (metrics.go:32-47).

2b. permswebhook — /Users/nodo/work/tasks/COR-985-go-nuts/entiredb/internal/entirecore/permswebhook/consumer.go

Constants (consumer.go:31-78): defaultAckWait = 2*time.Minute, defaultMaxDeliver = -1 (unlimited), defaultNakDelay = 5*time.Second, defaultFetchBatch = 1, defaultRetryBackoffMax = 30*time.Second, defaultInactiveThreshold = 72*time.Hour.

Delay formula — nakDelay(numDelivered uint64) (consumer.go:363-376): identical geometric ×2 shape to repoops, but capped at the configurable c.cfg.RetryBackoffMx (default 30s) instead of a package const:

deliveryCount at consumer.go:381-387 (identical helper). Comment at consumer.go:358-362 says it "mirrors repoops' backoff".

MaxDeliver handling: default -1, set into jsconsumer.Config (consumer.go:182-193, includes InactiveThreshold).

Disposition logic — disposeError (consumer.go:328-356): chooses Term vs Nak (no DLQ in permswebhook):

• Permanent errors are wrapped via permanent() / permanentError (consumer.go:389-401); isPermanent gates Term.
• terminate (consumer.go:285-300) and recoverPanic (consumer.go:307-323) both call msg.Term() directly. No DLQ publish and no MaxDeliver-based final-delivery handling — permswebhook disposes purely with Ack/Nak(delay)/Term.

Migration note: both consumers' hand-rolled nakDelay + disposition map cleanly onto go-nuts backoff.Policy (backoff/backoff.go): Policy{NakDelay, Factor: 2, MaxDelay: <cap>, MaxDeliver, TermOnExhaustion} with NakOrTerm(msg) / DelayFor(n) / IsFinalDelivery(msg, maxDeliver). The << shift geometric backoff exists only in these two files (repoops/consumer.go:856, permswebhook/consumer.go:371) — no other consumer computes redelivery delay.

3. All NATS publishers — legacy vs modern, hand-rolled tracing

Every publisher below is legacy (nats.JetStreamContext, i.e. PublishMsg(m *nats.Msg, opts ...nats.PubOpt) (*nats.PubAck, error)). No publisher in the repo uses the modern jetstream.JetStream for publishing, and none uses go-nuts natsmsg.Publisher yet (grep of natsmsg. symbols shows only DeadLetter, KeepInProgress, DLQPublisher, ClampToInt64, HeaderCarrier, ExtractHeader, SubjectToken — never natsmsg.Publisher/.Publish(). The only modern jetstream.JetStream usages are test/consumer-side (test/testutils/natstest.go:21, permssweep/consumer.go:121).

Legacy JetStream publishers (candidates for the natsmsg legacy adapter):

1. repoops — internal/entirecore/repoops/publisher.go

   • type JetStreamPublisher interface { PublishMsg(...) } (:30-34); func (p *Publisher) PublishTeardown/PublishAppSuspend/PublishResume → publishOp(ctx, ...) (:55-131).
   • Hand-rolled: own producer span, Nats-Msg-Id (NatsMsgIDHeader const :28), and its own headerCarrier trace-injection adapter (:133-152). Strong candidate.

2. gitjobs — core/mirrorrepo/gitjobs/publisher.go

   • type JetStreamPublisher interface {...} (:29-31); func (p *Publisher) PublishSyncRepo (:57-129).
   • Hand-rolled span + NatsMsgIDHeader (:24) + its own exported HeaderCarrier (:131-153) — explicitly "reused by other producers".

3. resync — internal/entirecore/resync_publisher.go

   • type resyncWebhookPublisher struct { js gitjobs.JetStreamPublisher }; func (p *resyncWebhookPublisher) PublishResync (:65-124).
   • Hand-rolled span; reuses gitjobs.NatsMsgIDHeader + gitjobs.HeaderCarrier (:100-101). Candidate.

4. permssweep — internal/entirecore/permssweep/publisher.go

   • type JetStreamPublisher interface {...} (:35-37); func (p *Publisher) PublishSweep (:72) and PublishCanary (:120), shared tail publishMsg (:58-65).
   • Already imports go-nuts natsmsg and uses natsmsg.HeaderCarrier (:61) + natsmsg.ClampToInt64 (:103,149), but still injects/publishes by hand with a local natsMsgIDHeader const (:23) and its own 15s publishTimeout (:31). Candidate.

5. permsbackfill — internal/entirecore/permsbackfill/publisher.go

   • type JetStreamPublisher interface {...} (:26-28); func (p *Publisher) PublishBackfill (:47-95).
   • Uses natsmsg.HeaderCarrier (:75) + natsmsg.ClampToInt64 (:85) but local natsMsgIDHeader const (:22) and hand-rolled span. Candidate.

6. email outbox relay — core/email/outbox.go

   • type Publisher interface { PublishMsg(m *nats.Msg, opts ...nats.PubOpt) (*nats.PubAck, error) } (:35-39); func (r *Relay) DrainOnce publishes at :180 with nats.MsgIdHdr set at :179. No span/trace injection here. (The email consumer is already migrated — see §6.) Candidate.

7. refevents — refevents/refevents.go + refevents/outbox.go

   • refevents.go:107-110 type Publisher interface {...} ("subset of nats.JetStreamContext"); Emitter.Emit publishes at :167 (sets nats.MsgIdHdr :166).
   • outbox.go Relay.pub Publisher (:261), publishes at :485 (r.pub.PublishMsg(msg, nats.Context(spanCtx))). Candidate; best-effort/outbox pattern.

Core-NATS (not JetStream) publisher — likely NOT a legacy-adapter candidate:

1. resourceevents — core/resourceevents/resourceevents.go
   • type Publisher interface { PublishMsg(m *nats.Msg) error } (:215-217) — plain core NATS (*nats.Conn), no PubAck, at-most-once best-effort by design (Emitter.Publish at :239-299, publish at :289). Hand-rolled span + its own headerCarrier (:301-320). This is not a JetStream publisher, so the JetStream legacy adapter does not apply directly; flag for separate handling.

4. Remaining natsx (and deleted-helper) references

Full case-insensitive sweep found exactly 3 references, all in doc comments (no live code, no imports):

• /Users/nodo/work/tasks/COR-985-go-nuts/entiredb/internal/entirecore/repoops/consumer.go:335 — "…Run's own create-or-update (via natsx) reconciles it thereafter."
• /Users/nodo/work/tasks/COR-985-go-nuts/entiredb/internal/entirecore/repoops/consumer.go:354 — "The create-or-update / consume / reconnect-with-backoff mechanism lives in natsx;"
• /Users/nodo/work/tasks/COR-985-go-nuts/entiredb/internal/entirecore/permswebhook/consumer.go:56 — "…the reconnect-backoff role it also played under natsx is now owned by go-nuts/jsconsumer."

These are stale: the mechanism they describe now lives in go-nuts/jsconsumer (both files already call jsconsumer.Run). Lines 335 and 354 are now inaccurate (Run uses jsconsumer, not natsx). No other deleted local NATS helper package names surfaced (no natsutil, no local nats/ helper package imports).

5. Tests and how they run

Delay-sequence / backoff tests:

• internal/entirecore/repoops/consumer_test.go:321 TestNakDelayBacksOff — asserts nakDelay(1)==base, nakDelay(2)>base, nakDelay(1000)==nakDelayMax.
• internal/entirecore/permswebhook/consumer_test.go:53 TestNakDelayBacksOff — same shape, cap = c.cfg.RetryBackoffMx. Also TestNewDefaultsUnlimitedMaxDeliver (:24), TestNewDefaultsInactiveThreshold (:37).
• These directly exercise the hand-rolled nakDelay and will need to move onto backoff.Policy.DelayFor/NakOrTerm when migrated.

Poison / disposition / panic-recovery tests:

• permswebhook/poison_test.go — TestNewPermsConsumer_CapturesMirrorsEagerly (:23), TestHandleMessage_PanicIsRecovered (:75, asserts msg.Termed). Uses github.com/entireio/go-nuts/natsmsg/natsmsgtest FakeMsg (:12,93).
• repoops consumer_test.go uses a local stubDLQ satisfying natsmsg.DLQPublisher (:17-19,134,154,165); TestNewRequiresDeps (:170) checks nil-DLQ rejection. There is no dedicated DLQ publish-then-Ack ordering test in repoops — stubDLQ is a no-op; the ordering in disposePoison is untested.
• Other repoops unit tests: TestNewFilterAndDurablePerOp (:201), TestConsumerConfigMatchesBindContract (:288), TestExecuteOp{Teardown,Suspend,Resume} (+ gate variants), TestSupersededGate, metrics_test.go, stream_test.go, jobs_test.go, publisher_test.go (dedup-key/subject/gate/publish-error tests).

Embedded NATS (JetStream) tests — via shared helper test/testutils/natstest.go EmbeddedJetStream(t) (:21), which starts an in-process nats-server/v2 and connects with go-nuts nuts.Connect(..., nuts.WithoutTLS()) (:35), returning a modern jetstream.JetStream. Callers:

• internal/entirecore/permswebhook/inactive_threshold_e2e_test.go:22 TestInactiveThreshold_AppliedAndReconciledEndToEnd (uses jsconsumer import).
• internal/entirecore/mirrordone/e2e_test.go:42, internal/entirecore/permssweep/e2e_test.go:21/92/145/192.

How tests run — no Makefile; tasks are mise (mise.toml, tasks under mise-tasks/):

• Unit: mise-tasks/test/_default → go test $RACE $VERBOSE -timeout $TIMEOUT $packages (also runs go vet && go test per nested module).
• Integration: mise-tasks/test/integration → go test ... $(go list ./test/... | grep -v /test/simulation).
• Also mise-tasks/test/{simulation,bench,determinism,all,oidf-conformance}. Lint: mise run lint (mise-tasks/lint/*). CLAUDE.md documents go test -v ./path -run TestName and TEST_SEED for simulation.

6. Where go-nuts is already used (imports of github.com/entireio/go-nuts/...)

Root package (nuts "github.com/entireio/go-nuts") — connection/drain/TLS helpers:

• test/testutils/natstest.go:7 (nuts.Connect, nuts.WithoutTLS)
• internal/entireserver/ref_events_publisher.go:9 (nuts.TLSConfigFromFiles, nuts.Connect, WithName/WithTLSConfig/WithRetryOnFailedConnect)
• internal/entirecore/serve_mode_global.go:11 (nuts.Drain, TLSConfigFromFiles, Connect, DefaultDrainTimeout)
• internal/entirecore/perms_sweep_worker.go:12 (nuts.Drain)
• internal/ciwebhooks/serve.go:11 (nuts.Drain, Connect, TLSConfigFromFiles)

jsconsumer (consumer supervise loop, jsconsumer.Run/Config/Start):

• core/email/consumer.go:11; internal/entirecore/permswebhook/consumer.go:11; internal/entirecore/repoops/consumer.go:11; internal/entirecore/mirrordone/consumer.go:11; internal/entirecore/verifiedemailwebhook/consumer.go:11; internal/entirecore/permsbackfill/consumer.go:11; internal/ciwebhooks/dispatcher/dispatcher.go:26; internal/entirecore/permswebhook/inactive_threshold_e2e_test.go:7.

natsmsg (helpers: ExtractHeader, KeepInProgress, DeadLetter, DLQPublisher, SubjectToken, ClampToInt64, HeaderCarrier):

• core/email/consumer.go:12; internal/entirecore/permswebhook/consumer.go:12; internal/entirecore/repoops/consumer.go:12; internal/entirecore/mirrordone/consumer.go:12; internal/entirecore/verifiedemailwebhook/consumer.go:12; internal/entirecore/permssweep/{consumer.go:12,publisher.go:9}; internal/entirecore/permsbackfill/publisher.go:8.

natsmsg/natsmsgtest (test fakes):

• core/email/consumer_test.go:10; internal/entirecore/permswebhook/poison_test.go:12; internal/entirecore/mirrordone/consumer_test.go:10; internal/entirecore/permssweep/consumer_test.go:11; internal/entirecore/permsbackfill/consumer_test.go:8; internal/entirecore/verifiedemailwebhook/consumer_test.go:9; internal/ciwebhooks/dispatcher/handlemessage_test.go:7.

backoff — not imported anywhere yet (grep for go-nuts/backoff returned nothing). This is the package the ticket wants repoops/permswebhook to adopt; the v0.7.0 backoff.Policy API (NakDelay, Factor, MaxDelay, MaxDeliver, TermOnExhaustion, NakOrTerm, DelayFor, IsFinalDelivery) already matches the hand-rolled logic in §2.

Best-in-class migration reference already in the repo: core/email/consumer.go is the fully-migrated consumer model (uses jsconsumer.Run + natsmsg helpers, defaultMaxDeliver = 50 with poison Term via ErrPoison) — but note its publisher side (core/email/outbox.go Relay) is still legacy PublishMsg and is a §3 adapter candidate.</result> <usage><subagent_tokens>165955</subagent_tokens><tool_uses>38</tool_uses><duration_ms>307965</duration_ms></usage> </task-notification>

medium effort → 3+5 angles × 6 candidates → 1-vote verify → ≤8 findings

You are reviewing for precision at medium effort: every finding you surface should be one a maintainer would act on.

Phase 0 — Gather the diff

Run git diff @{upstream}...HEAD (or git diff main...HEAD / git diff HEAD~1 if there's no upstream) to get the unified diff under review. If there are uncommitted changes, or the range diff is empty, also run git diff HEAD and include the working-tree changes in scope — the review often runs before the commit. If a PR number, branch name, or file path was passed as an argument, review that target instead. Treat this diff as the review scope.

Phase 1 — Find candidates (3 correctness angles + 3 cleanup angles + 1 altitude angle + 1 conventions angle, up to 6 each)

Run 8 independent finder angles via the Agent tool. Each surfaces up to 6 candidate findings with file, line, a one-line summary, and a concrete failure_scenario.

Angle A — line-by-line diff scan

Read every hunk in the diff, line by line. Then Read the enclosing function for each hunk — bugs in unchanged lines of a touched function are in scope (the PR re-exposes or fails to fix them). For every line ask: what input, state, timing, or platform makes this line wrong? Look for inverted/wrong conditions, off-by-one, null/undefined deref, missing await, falsy-zero checks, wrong-variable copy-paste, error swallowed in catch, unescaped regex metachars.

Angle B — removed-behavior auditor

For every line the diff DELETES or replaces, name the invariant or behavior it enforced, then search the new code for where that invariant is re-established. If you can't find it, that's a candidate: a removed guard, a dropped error path, a narrowed validation, a deleted test that was covering a real case.

Angle C — cross-file tracer

For each function the diff changes, find its callers (Grep for the symbol) and check whether the change breaks any call site: a new precondition, a changed return shape, a new exception, a timing/ordering dependency. Also check callees: does a parallel change in the same PR make a call unsafe?

Reuse

The angles above hunt for bugs; this one and the next two hunt for cleanup in the changed code. Flag new code that re-implements something the codebase already has — Grep shared/utility modules and files adjacent to the change, and name the existing helper to call instead.

Simplification

Flag unnecessary complexity the diff adds: redundant or derivable state, copy-paste with slight variation, deep nesting, dead code left behind. Name the simpler form that does the same job.

Efficiency

Flag wasted work the diff introduces: redundant computation or repeated I/O, independent operations run sequentially, blocking work added to startup or hot paths. Also flag long-lived objects built from closures or captured environments — they keep the entire enclosing scope alive for the object's lifetime (a memory leak when that scope holds large values); prefer a class/struct that copies only the fields it needs. Name the cheaper alternative.

Altitude

Check that each change is implemented at the right depth, not as a fragile bandaid. Special cases layered on shared infrastructure are a sign the fix isn't deep enough — prefer generalizing the underlying mechanism over adding special cases.

Conventions (CLAUDE.md)

Find the CLAUDE.md files that govern the changed code: the user-level ~/.claude/CLAUDE.md, the repo-root CLAUDE.md, plus any CLAUDE.md or CLAUDE.local.md in a directory that is an ancestor of a changed file (a directory's CLAUDE.md only applies to files at or below it). Read each one that exists, then check the diff for clear violations of the rules they state.

Only flag a violation when you can quote the exact rule and the exact line that breaks it — no style preferences, no vague "spirit of the doc" inferences. In the finding, name the CLAUDE.md path and quote the rule so the report can cite it. If no CLAUDE.md applies, return nothing for this angle.

Cleanup, altitude, and conventions candidates use the same file/line/summary shape; in failure_scenario, state the concrete cost (what is duplicated, wasted, harder to maintain, or which CLAUDE.md rule is broken) instead of a crash. Correctness bugs always outrank cleanup, altitude, and conventions findings when the output cap forces a cut.

Pass every candidate with a nameable failure scenario through — finders that silently drop half-believed candidates bypass the verify step and are the dominant cause of misses.

Phase 2 — Verify (1-vote, 3-state)

Dedup candidates that point at the same line/mechanism, keeping the one with the most concrete failure scenario. For each remaining candidate, run one verifier via the Agent tool: give it the diff, the relevant file(s), and the candidate, and have it return exactly one of:

• CONFIRMED — can name the inputs/state that trigger it and the wrong output or crash. Quote the line.
• PLAUSIBLE — mechanism is real, trigger is uncertain (timing, env, config). State what would confirm it.
• REFUTED — factually wrong (code doesn't say that) or guarded elsewhere. Quote the line that proves it.

Keep candidates where the vote is CONFIRMED or PLAUSIBLE.

Output

Return findings as a JSON array of at most 8 objects:

Ranked most-severe first. If more than 8 survive, keep the 8 most severe. If nothing survives verification, return [].

Key verifications performed:

• Inject/Extract parity (6 sites): natsmsg.Inject/Extract are semantically identical to the deleted otelnats versions (same nil-header guards, same global-propagator source, header creation on Inject). natsmsg.HeaderCarrier additionally implements Values() (a baggage-fidelity improvement, not a regression). No non-vendor code references otelnats.HeaderCarrier in signatures/tests, so nothing broke. nats.MsgIdHdr == "Nats-Msg-Id" == the retained MsgIDHeader, so the dedup key is unchanged.
• natspub rewrite: standard 3 attrs + SpanAttrs keys/order identical; messaging.message.id, sequence, duplicate preserved; ClampToInt64 is behavior-preserving for realistic sequences; timeout derivation (WithTimeout(ctx, AckWait) + nats.Context) identical; log fields unchanged. All observed deltas match the documented list exactly (no marshal-error span, status text "publish", messaging.nats.stream added, extra wrap layer). natspub_test.go still uses otelcodes (line 182), so no unused-import break; updated assertions pass.
• Typed publishers: all pass non-empty MsgID; their trace tests assert only TraceID continuity (preserved, producer span is a child). webhookevents.classifyPublishError uses a substring match on "maximum bytes exceeded", which survives the extra wrap layer; the latency histogram wraps the same full natspub.Publish call. No test asserts the old span shape.
• Build/vendor: modules.txt lists go-nuts v0.8.0 + natsmsg; vendored natsmsg is byte-identical to source; no lingering otelnats import anywhere.
• Logging: no otel-slog bridge is vendored/used, so the "caller span_id" delta is not observable in log output — not a real finding.

The migration is clean and behavior-preserving. The only genuinely review-worthy residual risks are externally-facing (outside this repo's test coverage):

[ { "file": "/Users/nodo/work/tasks/COR-985-go-nuts/mirror-pipeline/internal/natspub/natspub.go", "line": 94, "summary": "Publish errors now carry an extra middle layer: publish <ErrKind>: natsmsg: publish <subject>: <cause> (go-nuts publish.go:212 wraps before natspub re-wraps), vs the old publish <ErrKind>: <cause>. In-repo consumers are safe — webhookevents.classifyPublishError matches the innermost substring "maximum bytes exceeded" and errors.Is chains still work — but the contiguous error text changed.", "failure_scenario": "An out-of-repo log/alert monitor keyed on the old contiguous error substring (e.g. Datadog matching "publish forwarded webhook: nats:", now split apart by the inserted "natsmsg: publish <subject>:") silently stops matching, masking a stream_full/backpressure incident that used to page." }, { "file": "/Users/nodo/work/tasks/COR-985-go-nuts/mirror-pipeline/internal/natspub/natspub.go", "line": 92, "summary": "Undocumented delta (not in the stated list): the old natspub unconditionally set the Nats-Msg-Id header and the messaging.message.id span attr; go-nuts natsmsg.publish (vendored publish.go:192-198) sets both only when msgID != "". All current callers pass a non-empty MsgID, so no impact today; empty-string dedup was already a no-op, so even the edge is functionally equivalent.", "failure_scenario": "A future caller (or a code path where the id-composing inputs are empty) passes an empty pub.MsgID: the new code emits no Nats-Msg-Id header and no messaging.message.id attribute, whereas the old code set an (empty) header — a silent divergence in header presence that a header-presence assertion or dedup-audit tooling could trip on." } ]</result> <usage><subagent_tokens>86316</subagent_tokens><tool_uses>23</tool_uses><duration_ms>345861</duration_ms></usage> </task-notification>

merged

is the legacy-bridge use anywhere?

we should migrate off the legacy publisher no?