feat(org): add durable agent activation delivery

[chocobar]

Summary

• Add a durable JetStream stream for org agent activations.
• Use one explicit-ack durable consumer per subscribed agent with one in-flight message, preserving per-agent FIFO and fan-out semantics.
• Retry failed activations with NAK and refresh long-running leases with InProgress.
• Restore durable consumers after API restart and stop them with the server lifecycle context.
• Add JetStream redelivery/restart tests and agent fan-out/retry tests.

Verification

• go test ./pkg/org/...
• go test ./pkg/org/infrastructure/agentdelivery ./pkg/org/application/dispatch ./pkg/pubsub ./pkg/server
• go build ./pkg/server/ ./pkg/store/ ./pkg/types/

Known follow-up

This PR does not close the Postgres event append -> JetStream publish crash window. Publishing still appends the event before dispatch; a JetStream outage after append can leave a persisted event without a durable activation record. A transactional outbox or reconciliation worker is required before claiming complete end-to-end reliable delivery.

[helixml-bot]

PR Review: feat(org): add durable agent activation delivery

Summary

This PR adds JetStream-backed durable delivery for agent activations, replacing the in-memory queue for production use. The design preserves per-agent FIFO ordering with independent fan-out semantics.

---

Architecture Assessment ✅

The layering is clean:

• pubsub.DurablePubSub – interface + NATS JetStream implementation
• agentdelivery.Queue – agent-specific queue semantics on top of durable pubsub
• dispatch.ActivationQueue – pluggable interface keeping Dispatcher decoupled

The one-consumer-per-agent model (MaxAckPending: 1) ensures strict FIFO per agent while allowing parallel delivery across agents. Good choice.

---

Strengths

1. Restart recovery – Start() re-hydrates all durable consumers from JetStream after API restart.
2. Lease refresh – The 10-minute InProgress() ticker prevents premature redelivery during long activations (ack wait is 30 min).
3. Retry with backoff – NakWithDelay(time.Second) handles transient failures without hot-looping.
4. Good test coverage – Tests cover retry, fan-out, NAK redelivery, and server-restart scenarios.
5. Transparent integration – Existing dispatch callers are unchanged; the durable queue is injected via RegisterActivationQueue.

---

Concerns & Suggestions

1. Silent publish failures in Enqueue

if err := q.pubsub.PublishDurable(q.ctx, streamName, subject, payload); err != nil {
    q.logger.Error("agent delivery: publish", "agent", agentID, "err", err)
}

The caller has no indication the activation was lost. Consider:

• Returning an error from Enqueue (breaking change), or
• Buffering failed publishes for retry, or
• At minimum, emitting a metric so operators can alert on drops.

2. Consumer name collision risk

consumerName hashes orgID + "\x00" + agentID with SHA-256 truncated to 16 bytes (128 bits). Collision probability is negligible for practical agent counts, but the truncation loses information. If agent IDs are already globally unique within an org, why not use hex(orgID[:8]) + "-" + hex(agentID[:8]) or similar deterministic naming that's debuggable? (Minor; current approach is fine for correctness.)

3. No cleanup of orphaned consumers

When an agent is deleted, its durable consumer remains in JetStream forever. Over time this could accumulate stale consumers. Consider:

• Deleting the consumer when the agent is removed from the org, or
• Adding an InactiveThreshold (the test asserts Zero, but a 30-day threshold would auto-prune).

4. WorkQueuePolicy semantics

With Retention: WorkQueuePolicy, messages are deleted immediately after ack. If a consumer is slow to start (e.g., API process crashed before Start()), unacked messages remain, which is correct. Just double-check that this matches intent vs InterestPolicy.

5. Missing Close() call at shutdown

durableQueue.Close() cancels the context, but I don't see it wired into server shutdown in helix_org.go. The context passed (ctx) is the route-registration context, not the server lifecycle context. If that context never cancels, consumers will leak on graceful shutdown. Wire durableQueue.Close() into a shutdown hook or use the server's lifecycle context.

6. Test helper visibility

stopTestNats references n.embeddedServer and n.conn, which are unexported fields. This works because the test is in the same package, but if you ever move tests to pubsub_test, it will break. Consider adding a Close() method on Nats that cleanly shuts down both.

---

Known Follow-up (from PR description)

The acknowledged crash window (Postgres append → JetStream publish) is real. A transactional outbox or reconciliation worker is the right fix. The PR honestly documents this limitation—good.

---

Verdict

Approve with minor suggestions. The implementation is solid and well-tested. Address the shutdown hook gap (concern #5) before merging; the others can be follow-up issues.

Nice work @chocobar 👍