feat(governed): enforce signed decisions on transfers

README.md

@@ -30,8 +30,9 @@ rt.Register(dataexchange.NewService(dataexchange.ServiceConfig{}))
 
 | File | What it does |
 |---|---|
-| `dataexchange.go` | Wire format: `Frame`, `WriteFrame`, `ReadFrame`, `TraceFrame`, `TypeText/Binary/JSON/File/Trace`, `TypeName`. |
+| `dataexchange.go` | Wire format: `Frame`, `WriteFrame`, `ReadFrame`, `TraceFrame`, `TypeText/Binary/JSON/File/Trace/Governed`, `TypeName`. |
 | `client.go` | `Client` — `Dial` and send helpers. |
+| `governed.go` | Signed decision envelope, receiver-side verifier, and enforceable transport constraints. |
 | `server.go` | `Server` — accept loop and handler dispatch. |
 | `service.go` | `*Service` — `coreapi.Service` adapter. Build tag `!no_dataexchange`. |
 | `service_disabled.go` | Stub `*Service` for `-tags no_dataexchange` builds. |
@@ -45,7 +46,92 @@ rt.Register(dataexchange.NewService(dataexchange.ServiceConfig{}))
 For `TypeFile` the payload is prefixed with `[2-byte name length][name bytes]`.
 For `TypeTrace` the payload is `[4-byte inner_type][8-byte sent_at_ns][inner payload]`.
 
-Max frame size: 256 MiB.
+Max frame size: 64 MiB by default (configurable at process start with
+`PILOT_DATAEXCHANGE_MAX_FRAME` within its documented safe range).
+
+## Governed delivery
+
+`TypeGoverned` wraps one text, JSON, binary, or single-frame file delivery in
+the sender's signed `decision.Intent` and signed `decision.Decision`. The
+intent payload hash binds the frame type, filename, and exact bytes; the
+receiver verifies the signatures, tenant authority state, local deterministic
+ceiling, exact local destination, and any applicable transport constraints
+before it writes to disk. A workflow-approved action uses the same short-lived
+execution Decision as an ordinary allowed action—there is no reusable
+transport permit.
+
+Large resumable files use `TypeGovernedFileStream`: the signed envelope binds
+the exact `TypeFileStream` INIT (filename, declared length, full SHA-256,
+chunk size, and transfer ID). Required receivers admit later chunks only for
+that verified transfer on the same connection. Compute the file hash first,
+call `BuildStreamInitPayload`, sign an Intent using
+`GovernedStreamPayloadHash`, then call `SendGovernedFileStream`.
+
+For a required typed-disclosure profile, build a `decision.DisclosureBinding`
+whose content hash, byte length, filename, and stream transfer ID match the
+file, bind its canonical hash in the signed Intent, and use
+`SendGovernedWithDisclosure` or `SendGovernedFileStreamWithDisclosure`. A
+`DecisionFrameVerifier` with `RequireDisclosure` rejects governed messages,
+single-frame files, and resumable stream INITs that omit this evidence.
+When a receipt recorder is configured, typed deliveries require its V2
+disclosure-evidence method; the resulting signed receipt binds the canonical
+disclosure hash without retaining the file or message body.
+
+Set `ServiceConfig.GovernedVerifier` to `DecisionFrameVerifier` (or an
+equivalent local verifier). Once all senders have been upgraded, set
+`ServiceConfig.RequireGoverned` to reject unsigned legacy deliveries. Roll out
+in that order: upgraded receiver with verification available, upgraded
+senders, then required mode. An older receiver does not understand
+`TypeGoverned`; a required receiver intentionally rejects `TypeTrace` and
+raw `TypeFileStream` INIT frames. Governed stream INITs are supported.
+
+For auditable enterprise ingress, configure `GovernedReceiptRecorder` and set
+`RequireGovernedReceipts`. The service writes the received message/file first,
+then requires the recorder to durably capture the exact signed Intent and
+Decision before it emits the success ACK or delivery event. If recording fails,
+the staged file is removed and the sender receives an error rather than a
+successful but unreceipted delivery.
+
+### Local content inspection
+
+`ServiceConfig.GovernedContentInspector` is an optional receiver-local hook
+that runs after the signed envelope and local policy ceiling are verified but
+before a message/file is released. Set
+`RequireGovernedContentInspection` to make startup fail unless the hook is
+present; an inspection error removes a staged file or rejects the message.
+`decision.PresidioInspector` is the included OSS adapter for bounded text,
+JSON, XML, YAML, and form content. It rejects unsupported binary/document
+types rather than truncating or silently skipping them. The inspector is local
+to the receiver; neither the decision authority nor the sender's authority
+receives plaintext for this check.
+
+Typed disclosure binding V2 adds a tenant-defined `retention_class` to the
+same Intent hash. A signed policy can select allowed classes; the receiver sees
+the bound metadata. Configure `GovernedRetentionPolicies` to map those classes
+to local expiry durations. The service writes an owner-only retention journal
+before a governed message/file becomes accepted (and before a streamed file's
+final rename), then removes the content after expiry across restarts. An
+unknown, V1, or unconfigured class is rejected when retention is enabled.
+This is deletion retention, not a legal-hold or WORM-storage implementation.
+
+### Per-agent transfer quotas
+
+`ServiceConfig.GovernedTransferQuota` admits a bounded number of bytes and/or
+actions for each signed `Intent.AgentID` in a fixed local window. It is charged
+only after governed verification, including the declared bytes of a verified
+stream INIT; a peer address cannot select or reset another agent's budget.
+Quota is deliberately charged for an admitted attempt even if later local DLP
+or receipt persistence rejects it, so repeatedly failing submissions cannot
+turn the scanner into an unmetered denial-of-service target.
+
+The v1 action mapping is:
+
+| Frame | Intent action |
+|---|---|
+| text | `data.send.text` |
+| JSON | `data.send.json` |
+| binary | `data.send.binary` |
+| file | `file.share` |
 
 ## Build tags
 

client.go

@@ -3,8 +3,11 @@
 package dataexchange
 
 import (
+	"fmt"
+	"io"
 	"time"
 
+	"github.com/pilot-protocol/common/decision"
 	"github.com/pilot-protocol/common/driver"
 	"github.com/pilot-protocol/common/protocol"
 )
@@ -43,6 +46,113 @@ func (c *Client) SendFile(name string, data []byte) error {
 	return WriteFrame(c.conn, &Frame{Type: TypeFile, Filename: name, Payload: data})
 }
 
+// SendGoverned sends a frame with exact signed intent/decision evidence.
+// The remote service must have RequireGoverned enabled to enforce it.
+func (c *Client) SendGoverned(frame *Frame, intent decision.Intent, result decision.Decision) error {
+	governed, err := NewGovernedFrame(frame, intent, result)
+	if err != nil {
+		return err
+	}
+	envelope, err := EncodeGovernedFrame(governed)
+	if err != nil {
+		return err
+	}
+	return WriteFrame(c.conn, envelope)
+}
+
+// SendGovernedWithDisclosure sends a single governed message or file with a
+// typed disclosure binding. The Intent and Decision must already have been
+// obtained for the exact binding.
+func (c *Client) SendGovernedWithDisclosure(frame *Frame, intent decision.Intent, result decision.Decision, disclosure decision.DisclosureBinding) error {
+	governed, err := NewGovernedFrameWithDisclosure(frame, intent, result, disclosure)
+	if err != nil {
+		return err
+	}
+	envelope, err := EncodeGovernedFrame(governed)
+	if err != nil {
+		return err
+	}
+	return WriteFrame(c.conn, envelope)
+}
+
+// SendGovernedFileStream sends a large resumable file with a signed authority
+// decision bound to its exact INIT metadata. Callers compute the intent payload
+// hash with GovernedStreamPayloadHash over the deterministic INIT payload (use
+// BuildStreamInitPayload after calculating the file hash), then pass the same
+// signed intent and decision here. The receiver accepts chunks only for that
+// verified transfer ID and records its receipt on successful completion.
+func (c *Client) SendGovernedFileStream(name string, r io.ReadSeeker, size int64, intent decision.Intent, result decision.Decision, stepTimeout time.Duration) (*StreamResult, error) {
+	return c.SendGovernedFileStreamWithAuthorizer(name, r, size, func(_ []byte) (decision.Intent, decision.Decision, error) {
+		return intent, result, nil
+	}, stepTimeout)
+}
+
+// SendGovernedFileStreamWithDisclosure sends a governed resumable file whose
+// Intent and authority Decision bind the supplied typed disclosure metadata.
+// The metadata must match the final INIT (content hash, size, filename, and
+// transfer ID) exactly; callers can derive these with BuildStreamInitPayload.
+func (c *Client) SendGovernedFileStreamWithDisclosure(name string, r io.ReadSeeker, size int64, intent decision.Intent, result decision.Decision, disclosure decision.DisclosureBinding, stepTimeout time.Duration) (*StreamResult, error) {
+	return streamSendWithInit(c.conn, name, r, size, stepTimeout, func(id [transferIDLen]byte, declaredSize uint64, hash [32]byte, chunkSize uint32, filename string) (*Frame, error) {
+		init := encodeInit(id, declaredSize, hash, chunkSize, filename)
+		governed, err := NewGovernedStreamInitWithDisclosure(init, intent, result, disclosure)
+		if err != nil {
+			return nil, err
+		}
+		return EncodeGovernedStreamInit(governed)
+	})
+}
+
+// GovernedStreamAuthorizer receives the exact FileStream INIT payload after
+// its stable transfer ID and content hash have been calculated. It must return
+// a short-lived signed Intent and Decision bound to that exact payload. The
+// callback runs before any stream frame is written, so a deny or unavailable
+// authority cannot produce a partial transfer.
+type GovernedStreamAuthorizer func(initPayload []byte) (decision.Intent, decision.Decision, error)
+
+// GovernedStreamDisclosureAuthorizer returns typed disclosure evidence for
+// the exact INIT. Hosted federation clients use this after uploading the full
+// file content and before the first stream byte is released to the peer.
+type GovernedStreamDisclosureAuthorizer func(initPayload []byte) (decision.Intent, decision.Decision, decision.DisclosureBinding, error)
+
+// SendGovernedFileStreamWithAuthorizer obtains an exact signed decision only
+// after the resumable stream's INIT bytes are known. This is the preferred
+// sender API when the decision comes from an online authority.
+func (c *Client) SendGovernedFileStreamWithAuthorizer(name string, r io.ReadSeeker, size int64, authorize GovernedStreamAuthorizer, stepTimeout time.Duration) (*StreamResult, error) {
+	if authorize == nil {
+		return nil, fmt.Errorf("dataexchange: governed stream authorizer is required")
+	}
+	return streamSendWithInit(c.conn, name, r, size, stepTimeout, func(id [transferIDLen]byte, declaredSize uint64, hash [32]byte, chunkSize uint32, filename string) (*Frame, error) {
+		init := encodeInit(id, declaredSize, hash, chunkSize, filename)
+		intent, result, err := authorize(append([]byte(nil), init.Payload...))
+		if err != nil {
+			return nil, err
+		}
+		governed, err := NewGovernedStreamInit(init, intent, result)
+		if err != nil {
+			return nil, err
+		}
+		return EncodeGovernedStreamInit(governed)
+	})
+}
+
+func (c *Client) SendGovernedFileStreamWithDisclosureAuthorizer(name string, r io.ReadSeeker, size int64, authorize GovernedStreamDisclosureAuthorizer, stepTimeout time.Duration) (*StreamResult, error) {
+	if authorize == nil {
+		return nil, fmt.Errorf("dataexchange: governed stream disclosure authorizer is required")
+	}
+	return streamSendWithInit(c.conn, name, r, size, stepTimeout, func(id [transferIDLen]byte, declaredSize uint64, hash [32]byte, chunkSize uint32, filename string) (*Frame, error) {
+		init := encodeInit(id, declaredSize, hash, chunkSize, filename)
+		intent, result, disclosure, err := authorize(append([]byte(nil), init.Payload...))
+		if err != nil {
+			return nil, err
+		}
+		governed, err := NewGovernedStreamInitWithDisclosure(init, intent, result, disclosure)
+		if err != nil {
+			return nil, err
+		}
+		return EncodeGovernedStreamInit(governed)
+	})
+}
+
 // SendTrace wraps data in a TypeTrace frame with the current nanosecond clock.
 // Returns sentAtNs so the caller can correlate it against the timing ACK.
 func (c *Client) SendTrace(innerType uint32, data []byte) (sentAtNs int64, err error) {

dataexchange.go

@@ -33,6 +33,14 @@ const (
 	// peer that does not understand TypeFileStream never sends INIT-ACK, so
 	// the sender falls back to TypeFile.
 	TypeFileStream uint32 = 7
+	// TypeGoverned carries a regular data frame together with the sender's
+	// signed intent and authority decision. It is opt-in; enterprise receivers
+	// can require this envelope before persisting a message or file.
+	TypeGoverned uint32 = 8
+	// TypeGovernedFileStream carries the signed authorization evidence for a
+	// TypeFileStream INIT. Subsequent chunks are accepted only while bound to
+	// that verified transfer ID on the same connection.
+	TypeGovernedFileStream uint32 = 9
 )
 
 // TraceFrame carries timing metadata around an inner message frame.
@@ -253,6 +261,10 @@ func TypeName(t uint32) string {
 		return "TRACE"
 	case TypeFileStream:
 		return "FILESTREAM"
+	case TypeGoverned:
+		return "GOVERNED"
+	case TypeGovernedFileStream:
+		return "GOVERNED_FILESTREAM"
 	default:
 		return fmt.Sprintf("UNKNOWN(%d)", t)
 	}