feat(governed): enforce signed decisions on publications

README.md

@@ -49,8 +49,9 @@ pilotctl publish   <peer-address> ticker.btcusd --data '...'
 
 | File | What it does |
 |---|---|
-| `eventstream.go` | Wire format: `Event{Topic, Payload}`, length-prefixed framing. `WriteEvent` / `ReadEvent`. |
+| `eventstream.go` | Wire format: `Event{Topic, Payload}`, subscription policy interface, and length-prefixed framing. `WriteEvent` / `ReadEvent`. |
 | `client.go` | Subscriber side: `Client.Dial`, `.Subscribe(topic)`, `.Read`, `.Close`. |
+| `governed.go` | Signed publication envelope, broker-side verifier, and enforceable topic/payload constraints. |
 | `server.go` | Publisher side: `Server` accepts inbound stream connections and broadcasts. |
 | `service.go` | `*Service` — `coreapi.Service` adapter, binds port 1002. Build tag `!no_eventstream`. |
 | `service_disabled.go` | Stub when `-tags no_eventstream` is set. |
@@ -62,6 +63,58 @@ pilotctl publish   <peer-address> ticker.btcusd --data '...'
 |---|---|
 | `no_eventstream` | Compiles a no-op stub service. |
 
+## Governed publication
+
+An enterprise broker can call `SetGovernedPublication` before `Start` with a
+`DecisionEventVerifier` (or an equivalent local verifier). Its
+`GovernedTopic` transport envelope binds a publication topic and exact bytes to
+a signed `decision.Intent` and `decision.Decision`. The broker verifies the
+tenant authority state, local deterministic ceiling, exact broker resource,
+and understood constraints before it forwards the original event. Subscribers
+receive the original topic and payload, never the envelope.
+
+After all publishers have been upgraded, set `require` to `true`; unsigned
+legacy publications are then dropped at the broker. This is intentionally a
+publication control: subscription access remains governed by the existing
+`TopicPolicy`. A workflow-approved publication uses the same short-lived
+execution Decision as an ordinary allowed publication, rather than a reusable
+workflow token.
+
+For a typed-disclosure profile, use `PublishGovernedWithDisclosure` and bind
+the canonical `decision.DisclosureBinding` hash into the signed Intent. A
+`DecisionEventVerifier` with `RequireDisclosure` rejects otherwise-valid
+governed publications that do not carry matching content metadata.
+With a receipt recorder configured, typed publications require V2
+disclosure-evidence support before broker fanout; the signed receipt contains
+only the disclosure hash, not event plaintext.
+
+For auditable enterprise publishing, also call `SetGovernedReceiptRecorder`
+before `Start` and require it. The broker appends evidence for the exact signed
+Intent and Decision before fanout; a recorder failure denies the publication,
+so subscribers never receive an unreceipted governed event.
+
+### Local content inspection
+
+Call `SetGovernedContentInspector` before `Start` to inspect verified governed
+event bytes locally before broker fan-out. `RequireGovernedContentInspection`
+makes a missing hook a startup failure, while a detector error rejects only the
+publication. `decision.PresidioInspector` provides a bounded text/structured-
+text adapter for a tenant-local OSS Presidio service; unsupported binary types
+are rejected rather than bypassing inspection. This hook is not invoked by the
+central decision authority.
+
+Typed disclosure binding V2 can carry a signed `retention_class`, which a
+policy may restrict before fan-out. This binds metadata for downstream
+retention operations; it is not a substitute for a retention executor.
+
+### Per-agent publication quotas
+
+`SetGovernedTransferQuota` applies a bounded local byte/action budget to each
+verified publisher `Intent.AgentID`. The budget is charged after signature and
+local-policy verification, never from a network address or caller-supplied
+identity, and covers admitted attempts that later fail local inspection or
+receipt recording.
+
 ## License
 
 AGPL-3.0-or-later. See [LICENSE](LICENSE).

client.go

@@ -3,6 +3,7 @@
 package eventstream
 
 import (
+	"github.com/pilot-protocol/common/decision"
 	"github.com/pilot-protocol/common/driver"
 	"github.com/pilot-protocol/common/protocol"
 )
@@ -35,6 +36,35 @@ func (c *Client) Publish(topic string, payload []byte) error {
 	return WriteEvent(c.conn, &Event{Topic: topic, Payload: payload})
 }
 
+// PublishGoverned publishes an event with its exact signed intent and
+// authority decision. A broker configured to require governed publications
+// verifies the envelope before forwarding the inner event to subscribers.
+func (c *Client) PublishGoverned(event *Event, intent decision.Intent, result decision.Decision) error {
+	governed, err := NewGovernedEvent(event, intent, result)
+	if err != nil {
+		return err
+	}
+	envelope, err := EncodeGovernedEvent(governed)
+	if err != nil {
+		return err
+	}
+	return WriteEvent(c.conn, envelope)
+}
+
+// PublishGovernedWithDisclosure publishes a signed event whose Intent binds
+// typed disclosure metadata. The broker can require this form per topic.
+func (c *Client) PublishGovernedWithDisclosure(event *Event, intent decision.Intent, result decision.Decision, disclosure decision.DisclosureBinding) error {
+	governed, err := NewGovernedEventWithDisclosure(event, intent, result, disclosure)
+	if err != nil {
+		return err
+	}
+	envelope, err := EncodeGovernedEvent(governed)
+	if err != nil {
+		return err
+	}
+	return WriteEvent(c.conn, envelope)
+}
+
 // Recv waits for the next event from the broker.
 func (c *Client) Recv() (*Event, error) {
 	return ReadEvent(c.conn)

eventstream.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"io"
 	"unicode/utf8"
+
+	"github.com/pilot-protocol/common/coreapi"
 )
 
 // Event is a typed message published to the event stream.
@@ -16,6 +18,14 @@ type Event struct {
 	Payload []byte
 }
 
+// TopicPolicy is the authorization gate for topic subscription.
+// Implementations check whether a peer may subscribe to a named topic.
+// The enabled service uses an allow-all policy unless its caller installs a
+// stricter implementation.
+type TopicPolicy interface {
+	AllowSubscribe(remoteAddr coreapi.Addr, topic string) bool
+}
+
 // WriteEvent writes an event to a writer.
 func WriteEvent(w io.Writer, e *Event) error {
 	topic := []byte(e.Topic)

go.mod

@@ -2,4 +2,4 @@ module github.com/pilot-protocol/eventstream
 
 go 1.25.12
 
-require github.com/pilot-protocol/common v0.5.11
+require github.com/pilot-protocol/common v0.5.12

go.sum

@@ -1,2 +1,2 @@
-github.com/pilot-protocol/common v0.5.11 h1:gaPOT2v3/FUAx61lqPw7yRsrzmDSRWL4+LRuQh8mrqs=
-github.com/pilot-protocol/common v0.5.11/go.mod h1:Ybc6f1A37s3ShoEh1nBMVL9DPyYlxvkqPTvtbxaNWg4=
+github.com/pilot-protocol/common v0.5.12 h1:ZQ7v8oX0VYtEcluraQZvqrYPMvRIxjixSZtNz8Xo5Uc=
+github.com/pilot-protocol/common v0.5.12/go.mod h1:Ybc6f1A37s3ShoEh1nBMVL9DPyYlxvkqPTvtbxaNWg4=