feat(accounts): narrow authority through the policy-log witness, verified against a pinned key

configurations/local/policy-log-witness.env

@@ -0,0 +1 @@
+../../module/configurations/local/policy-log-witness.env

module/DATABASE_AUTHORITY.md

@@ -1105,18 +1105,39 @@ and deliberately holds no read or job-creation permission to recover its row.
 Those limitations are why the receipt issuer is a separate service rather than
 an accounts adapter that manufactures its own evidence.
 
-**This checkout does not yet have a runnable log transport.** The new service's
-witness, storage, signing-key loader, mirror, and private service declaration are
-implemented; protobuf and OpenAPI generation and the RPC listener remain blocked
-by the required generator prerequisite. Its README records that boundary and
-the remaining real-GCS qualification. The host's Go code and client wiring are
-unchanged, so this source addition must not be read as a running deployment or
-as completion of the policy-log condition.
-
-Until the transport and host client are wired, the existing host still serves
-normally and refuses the four narrowings above. Once wired, inability to reach
-the log expires the serving gate's freshness window and refuses service; an
-appended operation without a local commit remains a gap until reconciled.
+**The host's client is built; no environment composes it yet.**
+`infra.WitnessPolicyLog` implements the host's `PolicyLog` over the witness's
+private gRPC API. It verifies every receipt, record and cursor against the
+witness key pinned in the trust anchor (`policy_log_public_key.json`), compares
+the key the witness publishes with that pin without ever adopting it, refuses a
+receipt for bytes other than the ones offered, a sequence at or below one it
+has already verified, a broken chain link, and a cursor whose signed
+`issued_at` is outside the staleness window of the host's clock. The receipt it
+stores is the witness's own fields plus the verifying key id, so the row is
+evidence anyone holding the pin can re-verify offline. The witness holds an
+opaque entry — operation id, subject kind and decision, subject id, and the
+SHA-256 of the host's canonical encoding of the whole operation — never the
+policy or the actor.
+
+`work.go` wires it only when composition has delivered BOTH the witness endpoint
+and the pinned key; a delivered key that cannot be used refuses the boot. With
+either absent the host is configured without a witness: it serves normally and
+refuses the four narrowings above, and says so at startup. accounts declares no
+service dependency on `policy-log/grpc`: the endpoint is read from the
+`policy-log-witness` workspace group, which this module ships EMPTY and the
+composition root binds (`endpoint=${endpoint:saas/policy-log/grpc}`) only in an
+environment that provisions the witness. The edge, and the mesh admission the
+render derives from it, therefore exist only there; a local run never starts a
+witness it could not boot, and its host stays fail-closed. The dial is
+plaintext only under `internal-transport/mesh-protected=true` to an in-cluster
+Service address, and is refused by name otherwise.
+
+Once composed, inability to reach the witness expires the serving gate's
+freshness window and refuses service; an appended operation without a local
+commit remains a gap until reconciled. The real witness is exercised end to end
+by `TestTheRealWitnessWitnessesUninstallAndScopeRevocation`: an uninstall and a
+scope-grant revocation each commit the receipt the witness signed, and with the
+witness down both are refused and nothing narrows.
 
 ## Control-plane boundary
 

module/DEPLOYMENT_TOPOLOGY.md

@@ -32,11 +32,15 @@ The accounts descriptor catalog is an input: if its RPCs use gRPC, Connect, or
 REST without a corresponding accounts endpoint, generation fails.
 
 `policy-log` is in the inventory with one private `grpc` endpoint and appears in
-NO row of the service graph below, which is correct rather than an omission: it
-declares no dependency on anything, and the host's client for it is not built
-yet, so there is no edge to draw. When that client lands it adds an
-`accounts` → `policy-log/grpc` row and nothing else — the witness must never
-depend on the database whose authority it witnesses.
+NO row of the service graph below. It declares no dependency on anything, and
+accounts — whose witness client is built (`infra.WitnessPolicyLog`) — reaches it
+through the `policy-log-witness` workspace group rather than a declared edge.
+The module ships that group empty; the composition root binds
+`endpoint=${endpoint:saas/policy-log/grpc}` only in an environment that
+provisions the witness, and the render then admits `accounts` → `policy-log/grpc`
+there and nowhere else — so a local run never pulls in a witness that refuses
+to boot by its own contract. The witness must never depend on the database
+whose authority it witnesses.
 
 ## Service graph
 

module/configurations/local/policy-log-witness.env

@@ -0,0 +1,20 @@
+# Workspace group `policy-log-witness`: where this host reaches the policy-log
+# witness, as the composition root binds it.
+#
+# The edge to the witness exists only where the composition root binds it: an
+# environment that provisions the witness (its evidence store, its signing key
+# and the host's pinned public key) sets
+#
+#   endpoint=${endpoint:saas/policy-log/grpc}
+#
+# in its own copy of this group, and the render then both resolves the address
+# and admits this host as the witness endpoint's one consumer. This module does
+# not declare a service dependency on the witness, so a local run never starts
+# one: the witness refuses to boot without its bucket and its key, and a run
+# whose dependency refuses to boot ends.
+#
+# Empty here, and that is the local profile: no witness. The host then serves
+# and refuses every narrowing of authority, and says so at boot. Nothing locally
+# stands in for the witness, because a narrowing witnessed by a store with none
+# of the witness's guarantees would be a narrowing a restore can still undo.
+endpoint=

module/services/accounts/code/pkg/business/policy_log.go

@@ -192,9 +192,11 @@ type PolicyLog interface {
 
 	// Entries returns every entry after a sequence, in order.
 	//
-	// Used by reconciliation, never on a request path. The reader deliberately
-	// holds no job-creation permission on the log's warehouse, so this is a
-	// bounded row read with a restriction rather than a query.
+	// Used by reconciliation, never on a request path. A bounded page read
+	// from a sequence, never a query: the witness serves at most 500 records
+	// per read. Success is also what proves the log was REACHED, so an
+	// implementation must fail rather than return a page it could not verify
+	// is current.
 	Entries(ctx context.Context, afterSequence uint64, limit int) ([]*PolicyLogRecord, error)
 }
 
@@ -248,10 +250,10 @@ type PolicyLogGap struct {
 // log still holds its request, its connection and whatever the caller above it
 // holds. A bounded wait converts that into a refusal the caller can act on.
 //
-// Five seconds: long enough that an ordinary warehouse append succeeds, short
-// enough that an authority change fails visibly rather than hanging. A narrowing
-// that cannot be logged must not proceed, so the refusal is the correct outcome
-// and not a degraded one.
+// Five seconds: the witness's own append budget, long enough that an ordinary
+// append succeeds, short enough that an authority change fails visibly rather
+// than hanging. A narrowing that cannot be logged must not proceed, so the
+// refusal is the correct outcome and not a degraded one.
 const PolicyLogDeadline = 5 * time.Second
 
 // PolicyLogStaleAfter is how long a host may serve without having reached the
@@ -465,9 +467,9 @@ func (s *Service) ReconcilePolicyLog(ctx context.Context) (*PolicyLogServingStat
 	return state, err
 }
 
-// policyLogReadBatch bounds one reconciliation read. Bounded because the reader
-// holds no job-creation permission on the log's warehouse — this is a row read
-// with a restriction, and an unbounded one would be a scan.
+// policyLogReadBatch bounds one reconciliation read: the witness's own page
+// limit, so a pass reads at most one page and the next pass continues from the
+// cursor it advanced.
 const policyLogReadBatch = 500
 
 // policyNow is the host's clock, overridable in tests. Separate from the log's

module/services/accounts/code/pkg/business/policy_log_production_path_test.go

@@ -28,9 +28,11 @@ import (
 // `Service.UninstallSolution`, `WithPolicyLoggedNarrowing`, the receipts in
 // `policy_log_commits` through `infra.PostgresStore`, and `RequireServing`.
 //
-// The external log is the one collaborator that is a double, and it has to be:
-// it is external by design, it lives in a warehouse, and no deployment of it
-// exists (pkg/infra/warehouse_policy_log.go says why). A double for a
+// The external log is the one collaborator that is a double here, because these
+// tests make it misbehave in ways the real witness never would — refusing on
+// demand, failing a read — and assert what the host does about it. The real
+// witness is exercised separately, end to end, in
+// policy_log_witness_production_path_test.go. A double for a
 // collaborator is not a double for the thing under test — the mistake this PR
 // has already paid for twice is faking the PolicyLogStore, which IS the thing
 // under test, and that one is real postgres here.
@@ -39,7 +41,7 @@ import (
 // errors from real production statements given input they cannot accept, which
 // is why the gap they leave is a gap the real store wrote.
 
-// recordingPolicyLog stands in for the external warehouse log: it issues real
+// recordingPolicyLog stands in for the external witness: it issues real
 // monotonic sequences and remembers every entry, so a test can assert what the
 // host offered it and in what order.
 type recordingPolicyLog struct {

module/services/accounts/code/pkg/business/policy_log_witness_production_path_test.go

@@ -0,0 +1,195 @@
+//go:build !pure
+
+package business_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"accounts/pkg/business"
+	gen "accounts/pkg/gen/saas/accounts/v1"
+	policylogv1 "accounts/pkg/gen/saas/policylog/v1"
+	"accounts/pkg/infra"
+	"accounts/pkg/infra/policylogwitnesstest"
+	"accounts/pkg/infra/storetx"
+
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+// The policy log on the production path with the REAL witness: real narrowings,
+// through the real service, against real PostgreSQL, witnessed by the module's
+// own policy-log witness (its server package and witness log over its in-memory
+// object store, in a helper process — policylogwitnesstest says why it cannot
+// be linked here).
+//
+// Unlike policy_log_production_path_test.go, nothing on either side is a
+// double: the receipt each narrowing commits was sequenced, chained and signed
+// by the witness's code, and verified by this host's client under a key pinned
+// from the witness's own key document before the narrowing applied.
+//
+// The serving gate's reconciliation is not driven here. `policy_log_cursor` is
+// one row this package shares with the recording log every other test uses, and
+// pointing a reconciliation at a second log with its own sequence would — rightly
+// — be refused as a rollback. The gate's staleness behaviour is proven against
+// that log in the file next door.
+func TestTheRealWitnessWitnessesUninstallAndScopeRevocation(t *testing.T) {
+	clearData(t)
+	ctx := testCtx
+	closeEveryGap(t, ctx)
+	t.Cleanup(func() { closeEveryGap(t, ctx) })
+
+	witness := policylogwitnesstest.Start(t)
+	pin, err := infra.ParsePolicyLogWitnessKey(witness.KeyDocument)
+	require.NoError(t, err)
+	conn, err := grpc.NewClient(witness.Address, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = conn.Close() })
+	witnessLog, err := infra.NewWitnessPolicyLog(policylogv1.NewWitnessServiceClient(conn), pin)
+	require.NoError(t, err)
+
+	testService.SetPolicyLog(witnessLog, testStore)
+	t.Cleanup(func() { wirePolicyLogForTests(testService, testStore) })
+
+	adminID, orgID := mustUserAndOrg(t, ctx, "plwitness@example.com", "plwitness", "Witness Co")
+	memberID, _ := mustUserAndOrg(t, ctx, "plwitnessmember@example.com", "plwitnessmember", "Witness Member Co")
+	require.NoError(t, testService.AddOrgMember(ctx, adminID, &gen.AddOrgMemberRequest{
+		OrgId: orgID, UserId: memberID, Role: gen.OrgRole_ORG_ROLE_MEMBER,
+	}))
+	roleID := seedInstallableRole(t, ctx, orgID)
+	install := func(label string) string {
+		t.Helper()
+		targetID, _ := declarePresence(t, label)
+		installation, err := testService.InstallSolution(ctx, adminID, &business.InstallSolutionParams{
+			OrgID:           orgID,
+			AgentIdentifier: "acme.example/" + label + ":1.0.0",
+			TargetID:        targetID,
+			RootScopeLabel:  label,
+			RoleID:          roleID,
+			AllowedScopes:   []string{"doc"},
+		})
+		require.NoError(t, err)
+		return installation.Id
+	}
+	grant := func(scopePath string) *gen.RevokeScopeRequest {
+		t.Helper()
+		_, err := testService.RegisterScopeNode(ctx, adminID, &gen.RegisterScopeNodeRequest{
+			OrgId: orgID, ScopePath: scopePath, Kind: business.ScopeNodeKindSolution, Label: scopePath,
+		})
+		require.NoError(t, err)
+		granted, err := testService.GrantScope(ctx, adminID, &gen.GrantScopeRequest{
+			OrgId: orgID, SubjectId: memberID, SubjectKind: gen.SubjectKind_SUBJECT_KIND_PRINCIPAL,
+			ScopePath: scopePath, RoleId: roleID,
+		})
+		require.NoError(t, err)
+		return &gen.RevokeScopeRequest{
+			OrgId: orgID, SubjectId: memberID, SubjectKind: gen.SubjectKind_SUBJECT_KIND_PRINCIPAL,
+			ScopePath: granted.Grant.ScopePath, RoleId: granted.Grant.RoleId,
+		}
+	}
+
+	// --- The witness is up: both narrowings succeed, each with its receipt. ---
+
+	installed := install("plwitness-up")
+	require.NoError(t, testService.UninstallSolution(ctx, adminID, orgID, installed))
+	_, err = testStore.LiveModuleAuthority(ctx, business.ModulePrincipalID("plwitness-up"), orgID, installed)
+	require.ErrorIs(t, err, business.ErrModuleInstallationInactive, "the uninstall really applied")
+	uninstallRow := committedReceipt(t, ctx, business.PolicyLogInstallation, installed)
+
+	revoke := grant("plwitnessup")
+	require.Equal(t, 1, scopeGrantCount(t, ctx, orgID, memberID, revoke.ScopePath))
+	require.NoError(t, testService.RevokeScope(ctx, adminID, revoke))
+	require.Zero(t, scopeGrantCount(t, ctx, orgID, memberID, revoke.ScopePath), "the revocation really applied")
+	revokeRow := committedReceipt(t, ctx, business.PolicyLogScopeGrant, orgID+"/")
+
+	// Each committed receipt IS the witness's: a second replica reads the log
+	// back, verifying every signature and chain link, and finds both operations
+	// at the sequences, and with the receipts, this host committed.
+	reader, err := infra.NewWitnessPolicyLog(policylogv1.NewWitnessServiceClient(conn), pin)
+	require.NoError(t, err)
+	records, err := reader.Entries(ctx, 0, 500)
+	require.NoError(t, err)
+	require.Len(t, records, 2, "one entry per narrowing, and nothing else")
+	for _, row := range []receiptRow{uninstallRow, revokeRow} {
+		var found *business.PolicyLogRecord
+		for _, record := range records {
+			if record.OperationID == row.operationID {
+				found = record
+			}
+		}
+		require.NotNil(t, found, "operation %s reached the witness", row.operationID)
+		require.Equal(t, row.receipt, found.Receipt, "the stored receipt is the one the witness signed")
+		require.Equal(t, row.sequence, found.Sequence)
+		requireCommitted(t, ctx, row.operationID)
+	}
+
+	// --- The witness is down: both are refused, and nothing narrows. ---
+
+	stillInstalled := install("plwitness-down")
+	stillGranted := grant("plwitnessdown")
+	witness.Stop()
+
+	err = testService.UninstallSolution(ctx, adminID, orgID, stillInstalled)
+	require.ErrorIs(t, err, business.ErrPolicyLogUnreachable)
+	_, err = testStore.LiveModuleAuthority(ctx, business.ModulePrincipalID("plwitness-down"), orgID, stillInstalled)
+	require.NoError(t, err, "an unwitnessed uninstall must leave the installation live")
+
+	err = testService.RevokeScope(ctx, adminID, stillGranted)
+	require.ErrorIs(t, err, business.ErrPolicyLogUnreachable)
+	require.Equal(t, 1, scopeGrantCount(t, ctx, orgID, memberID, stillGranted.ScopePath),
+		"an unwitnessed revocation must leave the grant in place")
+
+	require.Zero(t, receiptCount(t, ctx, business.PolicyLogInstallation, stillInstalled),
+		"a refused append records no receipt, so no gap opens and nothing narrows silently")
+	require.Empty(t, uncommittedOperations(t, ctx))
+}
+
+type receiptRow struct {
+	operationID string
+	receipt     string
+	sequence    uint64
+}
+
+// committedReceipt reads the one committed policy_log_commits row for a subject
+// (or, for a subject id ending in "/", the one whose id starts with it).
+func committedReceipt(t *testing.T, ctx context.Context, kind business.PolicyLogSubjectKind, subject string) receiptRow {
+	t.Helper()
+	var row receiptRow
+	var sequence int64
+	var committedAt *time.Time
+	require.NoError(t, testStore.WithControlPlane(ctx, func(ctx context.Context) error {
+		return storetx.Tx(ctx).QueryRow(ctx, `
+			SELECT operation_id, receipt, log_sequence, committed_at
+			  FROM policy_log_commits
+			 WHERE subject_kind = $1
+			   AND (subject_id = $2 OR (right($2, 1) = '/' AND starts_with(subject_id, $2)))`,
+			string(kind), subject).Scan(&row.operationID, &row.receipt, &sequence, &committedAt)
+	}))
+	require.NotNil(t, committedAt, "the receipt committed in the narrowing's transaction")
+	row.sequence = uint64(sequence)
+	return row
+}
+
+func receiptCount(t *testing.T, ctx context.Context, kind business.PolicyLogSubjectKind, subject string) int {
+	t.Helper()
+	var count int
+	require.NoError(t, testStore.WithControlPlane(ctx, func(ctx context.Context) error {
+		return storetx.Tx(ctx).QueryRow(ctx,
+			`SELECT count(*) FROM policy_log_commits WHERE subject_kind = $1 AND subject_id = $2`,
+			string(kind), subject).Scan(&count)
+	}))
+	return count
+}
+
+func scopeGrantCount(t *testing.T, ctx context.Context, orgID, subjectID, scopePath string) int {
+	t.Helper()
+	var count int
+	require.NoError(t, testStore.WithControlPlane(ctx, func(ctx context.Context) error {
+		return storetx.Tx(ctx).QueryRow(ctx,
+			`SELECT count(*) FROM scope_grants WHERE org_id = $1 AND subject_id = $2 AND scope_path::text = $3`,
+			orgID, subjectID, scopePath).Scan(&count)
+	}))
+	return count
+}