[Hosted cutover 14/17] Add the fenced activation engine

.github/workflows/ci.yml

@@ -544,7 +544,7 @@ jobs:
 
       - name: Upload ${{ matrix.name }} test events
         if: failure()
-        uses: actions/upload-artifact@cf430e030ddbb5b0abf93d22962f4752f3646cd9  # v7.0.2
+        uses: actions/upload-artifact@v7.0.2
         with:
           name: go-test-events-windows-${{ matrix.name }}-attempt-${{ github.run_attempt }}
           path: go-test-events.jsonl

.gitignore

@@ -24,6 +24,7 @@ internal/web/dist/*
 /.sqlite-include/
 *.out
 coverage.out
+.hegel/
 *.test
 
 # Air (live-reload dev server)

cmd/agentsview/cli.go

@@ -708,7 +708,7 @@ func newActivityReportCommand() *cobra.Command {
 func newPGCommand() *cobra.Command {
 	return newReplicaCommand(
 		pgReplica{}, newPGVectorsCommand(), newPGServiceCommand(),
-		newPGHostedProvisionCommand(), newPGRawReparseCommand(),
+		newPGHostedProvisionCommand(), newPGRawReparseCommand(), newPGRawParityCommand(), newPGRawRolloutCommand(),
 	)
 }
 

cmd/agentsview/pg_raw_derive_pgtest_test.go

@@ -50,7 +50,7 @@ func hostedRuntimeConfig(t *testing.T) (config.Config, *sql.DB) {
 	password := make([]byte, 32)
 	_, err = rand.Read(password)
 	require.NoError(t, err)
-	_, err = admin.Exec(`CREATE ROLE "` + role + `" LOGIN NOSUPERUSER NOBYPASSRLS NOCREATEDB NOCREATEROLE NOINHERIT PASSWORD '` + hex.EncodeToString(password) + `';GRANT USAGE ON SCHEMA "` + schema + `" TO "` + role + `";GRANT SELECT,INSERT,UPDATE,DELETE ON ALL TABLES IN SCHEMA "` + schema + `" TO "` + role + `";GRANT USAGE ON ALL SEQUENCES IN SCHEMA "` + schema + `" TO "` + role + `"`)
+	_, err = admin.Exec(`CREATE ROLE "` + role + `" LOGIN NOSUPERUSER NOBYPASSRLS NOCREATEDB NOCREATEROLE NOINHERIT PASSWORD '` + hex.EncodeToString(password) + `';GRANT USAGE ON SCHEMA "` + schema + `" TO "` + role + `";GRANT SELECT,INSERT,UPDATE,DELETE ON ALL TABLES IN SCHEMA "` + schema + `" TO "` + role + `";GRANT USAGE ON ALL SEQUENCES IN SCHEMA "` + schema + `" TO "` + role + `";REVOKE INSERT,UPDATE,DELETE ON "` + schema + `".raw_rollout_scopes FROM "` + role + `"`)
 	require.NoError(t, err)
 	u, err := url.Parse(dsn)
 	require.NoError(t, err)

cmd/agentsview/pg_raw_parity.go

@@ -0,0 +1,81 @@
+package main
+
+import (
+	"database/sql"
+	"encoding/json/v2"
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/spf13/cobra"
+	"go.kenn.io/agentsview/internal/config"
+	"go.kenn.io/agentsview/internal/postgres"
+	"go.kenn.io/agentsview/internal/rawparity"
+)
+
+func newPGRawParityCommand() *cobra.Command {
+	var scope rawparity.Scope
+	cmd := &cobra.Command{Use: "raw-parity [target]", Short: "Compare hosted raw projections with retained pushed sessions", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error {
+		cfg, err := config.LoadMinimalFor(config.ScopePG)
+		if err != nil {
+			return err
+		}
+		target := ""
+		if len(args) > 0 {
+			target = args[0]
+		}
+		pg, err := cfg.ResolvePGTarget(target)
+		if err != nil {
+			return err
+		}
+		if pg.RawTenant == "" {
+			return errors.New("raw-parity requires raw_tenant in the selected PG target")
+		}
+		// OpenHosted uses the existing CheckHostedTenant runtime-role gate. No
+		// owner-mode bypass, schema provisioning, worker or local archive open.
+		database, err := postgres.OpenHosted(pg.URL, pg.Schema, pg.RawTenant, pg.AllowInsecure)
+		if err != nil {
+			return err
+		}
+		defer database.Close()
+		tx, err := database.BeginTx(cmd.Context(), &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true})
+		if err != nil {
+			return err
+		}
+		defer func() { _ = tx.Rollback() }()
+		report, err := rawparity.Run(cmd.Context(), postgres.RawParityReader{Tx: tx, Version: rawProcessingVersion()}, scope)
+		if err != nil {
+			return err
+		}
+		if err = tx.Commit(); err != nil {
+			return err
+		}
+		return writeRawParityReport(cmd.OutOrStdout(), cmd.ErrOrStderr(), report, outputFormat(cmd) == "json")
+	}}
+	cmd.Flags().StringVar(&scope.Device, "device", "", "Select a source device ID")
+	cmd.Flags().StringVar(&scope.Provider, "provider", "", "Select a provider")
+	registerFormatFlags(cmd.Flags())
+	return cmd
+}
+
+func writeRawParityReport(out, summary io.Writer, report rawparity.Report, asJSON bool) error {
+	if asJSON {
+		if err := json.MarshalWrite(out, report); err != nil {
+			return err
+		}
+		if _, err := fmt.Fprintln(out); err != nil {
+			return err
+		}
+	} else {
+		summary = out
+	}
+	projected, pending, failed, unsupported := 0, 0, 0, 0
+	for _, r := range report.Readiness {
+		projected += r.Projected
+		pending += r.Pending
+		failed += r.Failed
+		unsupported += r.Unsupported
+	}
+	_, err := fmt.Fprintf(summary, "parity pass=%t equal=%d differs=%d raw_only=%d legacy_only=%d ambiguous=%d unmatchable=%d projected=%d pending=%d failed=%d unsupported=%d\n", report.Pass, report.Counts["equal"], report.Counts["differs"], report.Counts["raw_only"], report.Counts["legacy_only"], report.Counts["ambiguous"], report.Counts["unmatchable"], projected, pending, failed, unsupported)
+	return err
+}

cmd/agentsview/pg_raw_parity_command_pgtest_test.go

@@ -0,0 +1,56 @@
+//go:build pgtest
+
+package main
+
+import (
+	"bytes"
+	"encoding/json/v2"
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.kenn.io/agentsview/internal/rawparity"
+)
+
+// No worker or subprocess is needed: even an empty hosted scope is a report,
+// not a command failure.
+func TestPGRawParityRuntimeCommand(t *testing.T) {
+	cfg, _ := hostedRuntimeConfig(t)
+	dir := testDataDir(t)
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "config.toml"), []byte(fmt.Sprintf("[pg.parity]\nurl = %q\nschema = %q\nraw_tenant = %q\n", cfg.PG.URL, cfg.PG.Schema, cfg.PG.RawTenant)), 0600))
+	for _, args := range [][]string{{"parity", "--json"}, {"parity", "--device", "missing", "--provider", "codex"}} {
+		cmd := newPGRawParityCommand()
+		var out, summary bytes.Buffer
+		cmd.SetOut(&out)
+		cmd.SetErr(&summary)
+		cmd.SetArgs(args)
+		require.NoError(t, cmd.Execute())
+		if len(args) == 2 {
+			var report rawparity.Report
+			require.NoError(t, json.Unmarshal(out.Bytes(), &report))
+			assert.False(t, report.Pass)
+			assert.Contains(t, summary.String(), "pass=false")
+		} else {
+			assert.Contains(t, out.String(), "pass=false")
+		}
+	}
+}
+
+func TestPGRawParityRejectsOwnerRole(t *testing.T) {
+	cfg, _ := hostedRuntimeConfig(t)
+	dir := testDataDir(t)
+	// Reuse only the isolated fixture's owner credentials, never a live target.
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "config.toml"), []byte(fmt.Sprintf("[pg.owner]\nurl = %q\nschema = %q\nraw_tenant = %q\n", os.Getenv("TEST_PG_URL"), cfg.PG.Schema, cfg.PG.RawTenant)), 0600))
+	cmd := newPGRawParityCommand()
+	var out, summary bytes.Buffer
+	cmd.SetOut(&out)
+	cmd.SetErr(&summary)
+	cmd.SetArgs([]string{"owner", "--json"})
+	cmd.SilenceErrors, cmd.SilenceUsage = true, true
+	require.ErrorContains(t, cmd.Execute(), "hosted runtime requires a restricted non-owner role")
+	assert.Empty(t, out.String())
+	assert.Empty(t, summary.String())
+}

cmd/agentsview/pg_raw_parity_test.go

@@ -0,0 +1,60 @@
+package main
+
+import (
+	"bytes"
+	"encoding/json/v2"
+	"errors"
+	"fmt"
+	"os"
+	"runtime"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.kenn.io/agentsview/internal/rawparity"
+	"hegel.dev/go/hegel"
+)
+
+// Counts and pass survive both output modes; JSON stdout stays machine-readable.
+func TestRawParityReportOutputProperty(t *testing.T) {
+	if runtime.GOOS == "darwin" && runtime.GOARCH == "amd64" && os.Getenv("HEGEL_LIBHEGEL_PATH") == "" {
+		t.Skip("Hegel requires HEGEL_LIBHEGEL_PATH on darwin/amd64")
+	}
+	hegel.Test(t, func(ht *hegel.T) {
+		n := hegel.Draw(ht, hegel.Integers(0, 100000))
+		r := rawparity.Report{Counts: rawparity.Counts{"equal": n}, Pass: n > 0}
+		var stdout, stderr bytes.Buffer
+		require.NoError(ht, writeRawParityReport(&stdout, &stderr, r, true))
+		var decoded rawparity.Report
+		require.NoError(ht, json.Unmarshal(stdout.Bytes(), &decoded))
+		assert.Equal(ht, r.Counts, decoded.Counts)
+		assert.Equal(ht, r.Pass, decoded.Pass)
+		assert.Contains(ht, stderr.String(), fmt.Sprintf("equal=%d", n))
+		assert.Contains(ht, stderr.String(), fmt.Sprintf("pass=%t", r.Pass))
+		stdout.Reset()
+		stderr.Reset()
+		require.NoError(ht, writeRawParityReport(&stdout, &stderr, r, false))
+		assert.Contains(ht, stdout.String(), fmt.Sprintf("equal=%d", n))
+		assert.Empty(ht, stderr.String())
+	}, hegel.WithTestCases(1000))
+}
+
+type parityFailWriter struct{ writes, failAt int }
+
+func (w *parityFailWriter) Write(p []byte) (int, error) {
+	w.writes++
+	if w.writes == w.failAt {
+		return 0, errors.New("output failed")
+	}
+	return len(p), nil
+}
+
+func TestRawParityWriterFailures(t *testing.T) {
+	for _, tc := range []struct {
+		json                 bool
+		outFail, summaryFail int
+	}{{true, 1, 0}, {true, 2, 0}, {true, 0, 1}, {false, 1, 0}} {
+		err := writeRawParityReport(&parityFailWriter{failAt: tc.outFail}, &parityFailWriter{failAt: tc.summaryFail}, rawparity.Report{}, tc.json)
+		assert.ErrorContains(t, err, "output failed")
+	}
+}

cmd/agentsview/pg_raw_rollout.go

@@ -0,0 +1,62 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/spf13/cobra"
+	"go.kenn.io/agentsview/internal/config"
+	"go.kenn.io/agentsview/internal/postgres"
+)
+
+func newPGRawRolloutCommand() *cobra.Command {
+	root := &cobra.Command{Use: "raw-rollout", Short: "Inspect or shadow hosted rollout scopes"}
+	for _, action := range []string{"list", "shadow", "clear"} {
+		var device, provider string
+		cmd := &cobra.Command{Use: action + " [target]", Args: cobra.MaximumNArgs(1), RunE: func(cmd *cobra.Command, args []string) error {
+			cfg, err := config.LoadMinimalFor(config.ScopePG)
+			if err != nil {
+				return err
+			}
+			applyClassifierConfig(cfg)
+			target := ""
+			if len(args) > 0 {
+				target = args[0]
+			}
+			pg, err := cfg.ResolvePGTarget(target)
+			if err != nil {
+				return err
+			}
+			if pg.RawTenant == "" {
+				return errors.New("raw-rollout requires raw_tenant in the selected owner target")
+			}
+			database, err := postgres.Open(pg.URL, pg.Schema, pg.AllowInsecure)
+			if err != nil {
+				return errors.New("opening hosted owner connection failed")
+			}
+			defer database.Close()
+			if action != "list" {
+				if err := postgres.ChangeRawRollout(cmd.Context(), database, pg.Schema, pg.RawTenant, device, provider, action); err != nil {
+					return err
+				}
+			}
+			scopes, err := postgres.ListRawRollout(cmd.Context(), database, pg.Schema, pg.RawTenant, device, provider)
+			if err != nil {
+				return err
+			}
+			fmt.Fprintln(cmd.OutOrStdout(), "DEVICE\tPROVIDER\tSTATE\tSERVED\tSHADOW\tSUPERSEDED\tPARITY")
+			for _, scope := range scopes {
+				parity := string(scope.Parity)
+				if parity == "" {
+					parity = "null"
+				}
+				fmt.Fprintf(cmd.OutOrStdout(), "%q\t%q\t%s\t%d\t%d\t%d\t%s\n", scope.DeviceID, scope.Provider, scope.State, scope.Served, scope.Shadow, scope.Superseded, parity)
+			}
+			return nil
+		}}
+		cmd.Flags().StringVar(&device, "device", "", "Select a device (empty means all)")
+		cmd.Flags().StringVar(&provider, "provider", "", "Select a provider (empty means all)")
+		root.AddCommand(cmd)
+	}
+	return root
+}

cmd/agentsview/pg_raw_rollout_pgtest_test.go

@@ -0,0 +1,34 @@
+//go:build pgtest
+
+package main
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRawRolloutCommandScopes(t *testing.T) {
+	cfg, _ := hostedRuntimeConfig(t)
+	t.Setenv("AGENTSVIEW_DATA_DIR", cfg.DataDir)
+	t.Setenv("AGENTSVIEW_PG_URL", "")
+	t.Setenv("AGENTSVIEW_PG_SCHEMA", "")
+	body := fmt.Sprintf("[pg.provision]\nurl = %q\nschema = %q\nraw_tenant = %q\n", os.Getenv("TEST_PG_URL"), cfg.PG.Schema, cfg.PG.RawTenant)
+	require.NoError(t, os.WriteFile(filepath.Join(cfg.DataDir, "config.toml"), []byte(body), 0600))
+	for _, args := range [][]string{{"shadow", "provision"}, {"shadow", "provision", "--device", "device-'alpha", "--provider", "codex"}} {
+		_, err := executeCommand(newPGRawRolloutCommand(), args...)
+		require.NoError(t, err)
+	}
+	output, err := executeCommand(newPGRawRolloutCommand(), "list", "provision", "--device", "device-'alpha")
+	require.NoError(t, err)
+	require.Contains(t, output, "\"device-'alpha\"\t\"codex\"\tshadow\t0\t0\t0\tnull")
+	_, err = executeCommand(newPGRawRolloutCommand(), "clear", "provision")
+	require.ErrorContains(t, err, "clear narrower rollout scopes")
+	for _, args := range [][]string{{"clear", "provision", "--device", "device-'alpha", "--provider", "codex"}, {"clear", "provision"}} {
+		_, err := executeCommand(newPGRawRolloutCommand(), args...)
+		require.NoError(t, err)
+	}
+}

docs/changelog.md

@@ -11,6 +11,25 @@ The latest release is
 
 **New features**
 
+- Operators can compare hosted raw sessions with retained pushed history using
+  `pg raw-parity`. Its JSON report names content differences, unresolved
+  identities and sources that are not ready, including trashed sessions.
+
+- Hosted operators can record tenant, device and provider rollout scopes and
+  inspect their physical session counts. Shadow sessions stay out of public
+  reads, search, reports and trash, while their stored data remains available
+  for comparison and rollback. Parity-gated activation is not available yet.
+
+- Hosted cutover comparisons resolve parent and subagent targets from each
+  source's physical evidence. Missing, excluded or ambiguous targets no longer
+  appear equal solely because stored aliases match. Hidden and trashed
+  targets remain part of the comparison.
+
+- Hosted cutover tooling can read complete session content and source readiness
+  in one read-only PostgreSQL snapshot, including trashed sessions, and match
+  sessions by recorded provider identity, keeping unresolved and source-local
+  collisions separate.
+
 - `agentsview agents` lists every supported agent, where AgentsView looks for its
   sessions, and which session details it reads, such as tool calls, thinking,
   subagents, and token usage. Add `--json` for scripts. It works without a