adapters/nats: optional NATS bridge for ai/verify barrier decisions

.github/workflows/ci.yml

@@ -185,11 +185,17 @@ jobs:
       # gap. Verified via GOOS=windows cross-compile + vet (including test
       # files) before adding this, given tonight's pattern of Windows-only
       # surprises in newly-covered packages.
+      #
+      # adapters/nats joined this loop later: another Go module under
+      # go.work, same reasoning as the five above. Its tests start an
+      # in-process NATS server (nats-server/v2's server package) rather
+      # than requiring one listening externally, so they run the same way
+      # here as they do locally, no new service dependency for this job.
       - name: Build, vet, and test the other workspace modules
         shell: bash
         run: |
           set -euo pipefail
-          for m in search jsondb incfs adapters/cassandra adapters/redis; do
+          for m in search jsondb incfs adapters/cassandra adapters/redis adapters/nats; do
             echo "::group::$m"
             (cd "$m" && go build ./... && go vet ./... && go test -timeout 20m ./...)
             echo "::endgroup::"

.github/workflows/security.yml

@@ -181,7 +181,7 @@ jobs:
           limit-severities-for-sarif: 'true'
 
       - name: Upload SARIF (critical/high)
-        if: always()
+        if: always() && hashFiles('trivy-image-critical-high.sarif') != ''
         uses: github/codeql-action/upload-sarif@v4
         with:
           sarif_file: trivy-image-critical-high.sarif

Dockerfile

@@ -15,6 +15,7 @@ WORKDIR /src
 COPY go.work go.work.sum* ./
 COPY go.mod go.sum ./
 COPY adapters/cassandra/go.mod adapters/cassandra/go.sum* ./adapters/cassandra/
+COPY adapters/nats/go.mod adapters/nats/go.sum* ./adapters/nats/
 COPY adapters/redis/go.mod adapters/redis/go.sum* ./adapters/redis/
 COPY ai/go.mod ai/go.sum* ./ai/
 COPY incfs/go.mod incfs/go.sum* ./incfs/

README.md

@@ -336,6 +336,7 @@ Every architecture involves tradeoffs. Here is an honest comparison of where Jol
 - **PostgreSQL**: Industry standard for general relational databases. Choose Postgres when you need complex relational schemas, advanced SQL aggregations, or standard ecosystem tooling. Joltrin is better suited when you want an embedded storage engine inside your application process without database server management.
 - **Redis**: Industry standard for ultra-low-latency in-memory key-value caching. Choose Redis when all data fits in RAM and you need simple cache operations. Joltrin provides durable B-Tree disk persistence, multi-account ACID transactions, and erasure coding.
 - **Kafka / RabbitMQ**: Industry standards for high-volume streaming and pub/sub. Choose Kafka when you need multi-datacenter event streams and log retention. Joltrin provides transactional task queues co-located with storage state for local swarms.
+- **NATS (optional, `adapters/nats`)**: not a replacement for anything joltrin embeds, and not on the hot path. If a team already runs NATS as part of their own architecture, `adapters/nats.VerifyBridge` will publish `ai/verify` barrier decisions to it, fire-and-forget, after the decision is already made, so another service outside joltrin's process can observe it without polling. Nothing imports this by default and a publish failure can never change the barrier's own answer. See the addendum in `docs/MCP_A2A_AND_VERIFICATION_ENGINE.md` for the full reasoning on why this doesn't reverse the embedded design.
 - **Temporal**: Industry standard for long-running durable workflows spanning external microservices. Choose Temporal for multi-week human-in-the-loop workflows across disparate clouds. Joltrin is designed for local-to-cluster co-located data and task execution.
 - **SQLite**: Industry standard for embedded single-file relational databases. Choose SQLite for client desktop/mobile apps needing SQL. Joltrin is designed for high-concurrency multi-threaded workers, clustered coordination, partitioned vector stores, and erasure coding.
 

adapters/nats/bridge.go

@@ -0,0 +1,170 @@
+package nats
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"github.com/sharedcode/joltrin/ai/verify"
+)
+
+// BarrierDecision is the structured event VerifyBridge publishes to NATS
+// every time it runs a barrier check. It carries enough detail for an
+// external subscriber to reconstruct what happened without re-deriving it
+// from the workflow graph: which step, which workflow, whether it was
+// allowed, and if not, which rule and which missing state blocked it.
+type BarrierDecision struct {
+	Time     time.Time     `json:"time"`
+	Workflow string        `json:"workflow,omitempty"`
+	Step     verify.StepID `json:"step"`
+	// Allowed is true when the step passed the barrier (CheckSafety
+	// returned nil). It is false both for a blocked *verify.Violation and
+	// for a malformed request (an unknown step), Rule/MissingState
+	// distinguish the two: a Violation always sets Rule, a malformed
+	// request never does.
+	Allowed bool `json:"allowed"`
+	// Replayed is true when the decision came from
+	// CheckAndCommitIdempotent's idempotency cache rather than a fresh
+	// check, i.e. this event describes a retried call, not a new one.
+	Replayed bool `json:"replayed,omitempty"`
+	// Rule is the SafetyRule.Name that blocked this step, or the literal
+	// "precondition" when the step's own precondition was not met. Empty
+	// when Allowed is true or the request was malformed.
+	Rule string `json:"rule,omitempty"`
+	// MissingState is the State that still needs to be established before
+	// this step can pass, taken from verify.Violation.MissingState. Empty
+	// when Allowed is true or the request was malformed.
+	MissingState verify.State `json:"missing_state,omitempty"`
+	// Reason is the blocking error's message, verbatim, for a human
+	// reading the event directly. Empty when Allowed is true.
+	Reason string `json:"reason,omitempty"`
+}
+
+// DefaultSubject is the NATS subject a VerifyBridge publishes to when
+// constructed with an empty workflow name.
+const DefaultSubject = "joltrin.verify.decision"
+
+// SubjectFor returns the NATS subject a VerifyBridge publishes
+// BarrierDecision events to for the given workflow name. An empty name
+// returns DefaultSubject.
+func SubjectFor(workflow string) string {
+	if workflow == "" {
+		return DefaultSubject
+	}
+	return fmt.Sprintf("joltrin.verify.%s.decision", workflow)
+}
+
+// VerifyBridge wraps a *verify.Workflow and publishes a BarrierDecision
+// event to NATS after every barrier check it runs. It does not change
+// ai/verify's behavior: CheckSafety, CheckAndCommit, and
+// CheckAndCommitIdempotent each call straight through to the identically
+// named *verify.Workflow method and return its exact result unchanged; the
+// NATS publish is a side effect on the way out. A publish failure is never
+// allowed to change or block the barrier's own decision, see publish
+// below.
+//
+// A VerifyBridge is only ever a decorator a caller opts into at its own
+// call sites. Nothing in ai/verify, tools/mcpserver, or tools/a2aagent
+// constructs or requires one.
+type VerifyBridge struct {
+	wf       *verify.Workflow
+	nc       *nats.Conn
+	workflow string
+	subject  string
+
+	// onPublishError, if set, is called with any error returned by the
+	// underlying NATS publish. Optional: a VerifyBridge with no handler set
+	// simply drops a publish failure, exactly like fire-and-forget metrics
+	// or logging, rather than letting it surface as a barrier error.
+	onPublishError func(error)
+}
+
+// NewVerifyBridge returns a VerifyBridge that runs barrier checks against
+// wf and publishes a BarrierDecision for each one to nc, on the subject
+// SubjectFor(workflowName). workflowName is a label only, used to build the
+// subject and to tag published events; it does not have to match any name
+// wf is registered under in a tools/runbookstore.Store.
+func NewVerifyBridge(wf *verify.Workflow, nc *nats.Conn, workflowName string) *VerifyBridge {
+	return &VerifyBridge{
+		wf:       wf,
+		nc:       nc,
+		workflow: workflowName,
+		subject:  SubjectFor(workflowName),
+	}
+}
+
+// OnPublishError sets a handler called with any error the underlying NATS
+// publish returns, and returns the bridge for chaining. Optional; without
+// one, a publish failure is dropped silently.
+func (b *VerifyBridge) OnPublishError(fn func(error)) *VerifyBridge {
+	b.onPublishError = fn
+	return b
+}
+
+// Subject returns the NATS subject this bridge publishes to.
+func (b *VerifyBridge) Subject() string {
+	return b.subject
+}
+
+// CheckSafety runs wf.CheckSafety(trace, next), publishes the resulting
+// BarrierDecision, and returns wf.CheckSafety's exact result.
+func (b *VerifyBridge) CheckSafety(trace *verify.Trace, next verify.StepID) error {
+	err := b.wf.CheckSafety(trace, next)
+	b.publish(next, false, err)
+	return err
+}
+
+// CheckAndCommit runs wf.CheckAndCommit(trace, next), publishes the
+// resulting BarrierDecision, and returns wf.CheckAndCommit's exact result.
+func (b *VerifyBridge) CheckAndCommit(trace *verify.Trace, next verify.StepID) error {
+	err := b.wf.CheckAndCommit(trace, next)
+	b.publish(next, false, err)
+	return err
+}
+
+// CheckAndCommitIdempotent runs
+// wf.CheckAndCommitIdempotent(trace, next, idempotencyKey), publishes the
+// resulting BarrierDecision (with Replayed set from the same call), and
+// returns wf.CheckAndCommitIdempotent's exact result.
+func (b *VerifyBridge) CheckAndCommitIdempotent(trace *verify.Trace, next verify.StepID, idempotencyKey string) (replayed bool, err error) {
+	replayed, err = b.wf.CheckAndCommitIdempotent(trace, next, idempotencyKey)
+	b.publish(next, replayed, err)
+	return replayed, err
+}
+
+func (b *VerifyBridge) publish(step verify.StepID, replayed bool, err error) {
+	ev := BarrierDecision{
+		Time:     time.Now().UTC(),
+		Workflow: b.workflow,
+		Step:     step,
+		Allowed:  err == nil,
+		Replayed: replayed,
+	}
+	if err != nil {
+		ev.Reason = err.Error()
+		var violation *verify.Violation
+		if errors.As(err, &violation) {
+			ev.Rule = violation.Rule
+			ev.MissingState = violation.MissingState
+		}
+	}
+
+	data, err := json.Marshal(ev)
+	if err != nil {
+		// A BarrierDecision is a fixed, JSON-safe shape; this would only
+		// fail if that shape changes to include something unmarshalable.
+		// Report it the same way a publish failure is reported, rather than
+		// silently dropping a bug.
+		if b.onPublishError != nil {
+			b.onPublishError(fmt.Errorf("nats: marshal barrier decision: %w", err))
+		}
+		return
+	}
+
+	if err := b.nc.Publish(b.subject, data); err != nil && b.onPublishError != nil {
+		b.onPublishError(fmt.Errorf("nats: publish barrier decision: %w", err))
+	}
+}