feat(g01): connect paired terminal execution to bounded broker handoff

docs/CI.md

@@ -31,19 +31,31 @@ Individual commands are available when iterating:
 | `make fuzz-smoke` | Run each discovered fuzz target for a fixed one-second smoke window, or print an explicit `SKIPPED` result when no target exists. |
 | `make deps` | Require a clean `go mod tidy -diff`, verified module sums and a read-only dependency load. |
 | `make licenses` | Compare the exact runtime module/version/replacement graph with its inventory and require a top-level license file. |
-| `make experiments` | Require both established G01/G02 modules, run their default race/vet suites, then exercise the two explicitly reviewed G01 CLI packages with `g01_live,g01_worker` tags and the reviewed `g01_pair_fixture` livecanary collection/listener and terminal partitions with tagged vet. |
+| `make experiments` | Require both established G01/G02 modules, run their static-partitioned default race/vet suites, then exercise the two explicitly reviewed G01 CLI packages with `g01_live,g01_worker` tags and the reviewed `g01_pair_fixture` livecanary collection/listener and terminal partitions with tagged vet. |
 | `make vuln` | Run the exact `golang.org/x/vuln/cmd/govulncheck@v1.7.0` tool. |
 
 No hardware, live GitHub, Docker or daemon suite is part of this public check. Those profiles remain explicit future or maintainer-controlled runs; they are not silently converted into passing tests here. G04 introduces the first application behavior contracts and should add meaningful unit and fuzz targets before claiming those forms of coverage.
 
-The default untagged G01 race suite keeps its existing 45-second per-process
-deadline while using two sequential, static partitions. The first runs the exact
-`TestBaselineStatisticsPresenceAndEligibility` name through `./...`; the second
-runs an unfiltered `./...` with only that exact name skipped. Keeping package
-discovery in both commands means a same-named test in another package is run in
-the first partition rather than silently dropped by a global skip, while the
-unfiltered remainder still executes every ordinary test, Example Output and fuzz
-seed. G02 retains its single default race invocation.
+The default untagged G01 and G02 race suites keep their existing 45-second
+per-process deadline. G01 uses two sequential static partitions: the exact
+`TestBaselineStatisticsPresenceAndEligibility` name through `./...`, then an
+unfiltered `./...` with only that exact name skipped. G02 uses three sequential
+static partitions, all with `-race -count=1 -timeout=45s` and `./...` package
+discovery: the exact `TestPairedBrokerRealCadenceChildExceedsThirtySeconds`
+name; the remaining `^TestPaired` family with that exact cadence name skipped;
+and an unfiltered complement that skips `^TestPaired`. Keeping package
+discovery in every command means a same-named test in another package is run in
+the matching named partition rather than silently dropped by a global skip,
+while the unfiltered G02 complement still executes every ordinary non-paired
+test, Example Output and fuzz seed. No widened named-test timeout is part of
+the public contract.
+
+The tooling regression matrix generates positive and independent failing
+witnesses for each G02 partition boundary: the named cadence test, the remaining
+`TestPaired` family, remainder, another package, a same-name cadence test in
+another package, a same-name remaining `TestPaired` test in another package, an
+Example Output and a fuzz seed. Each witness must execute exactly once, and
+each failing witness must propagate a nonzero offline-gate result.
 
 The tagged CLI tests use synthetic input/subprocess fixtures and static plan or
 refusal paths. The `g01_pair_fixture` livecanary checks use private synthetic

experiments/g01-scaleset/cmd/g01-live/fixture_support.go

@@ -0,0 +1,85 @@
+//go:build g01_live && g01_pair_fixture
+
+package main
+
+import (
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"syscall"
+
+	"github.com/1XP-AI/gh-runnerd/experiments/g01-scaleset/livecanary"
+	"github.com/1XP-AI/gh-runnerd/experiments/g01-scaleset/liveworker"
+)
+
+type fixtureEndpointConfig struct {
+	BaseURL            string `json:"base_url"`
+	CAPEM              string `json:"ca_pem"`
+	AdmissionDirectory string `json:"admission_directory"`
+}
+
+func readFixtureEndpointConfig(stateDirectory string) (fixtureEndpointConfig, error) {
+	var config fixtureEndpointConfig
+	path := filepath.Join(stateDirectory, "paired-fixture.json")
+	file, err := os.OpenFile(path, os.O_RDONLY|syscall.O_NOFOLLOW, 0)
+	if err != nil {
+		return config, livecanary.ErrApproval
+	}
+	defer file.Close()
+	info, err := file.Stat()
+	stat, ok := info.Sys().(*syscall.Stat_t)
+	if err != nil || !ok || int(stat.Uid) != os.Geteuid() || info.Mode().Perm() != 0600 || !info.Mode().IsRegular() || stat.Nlink != 1 || info.Size() > 8192 {
+		return config, livecanary.ErrApproval
+	}
+	data, err := io.ReadAll(io.LimitReader(file, 8193))
+	if err != nil || len(data) > 8192 || livecanary.DecodeStrict(data, &config) != nil || config.BaseURL == "" || config.CAPEM == "" || config.AdmissionDirectory == "" || !filepath.IsAbs(config.AdmissionDirectory) || filepath.Clean(config.AdmissionDirectory) != config.AdmissionDirectory {
+		return fixtureEndpointConfig{}, livecanary.ErrApproval
+	}
+	return config, nil
+}
+
+func init() {
+	openJournalForCommand = func(stateDirectory string, a livecanary.Approval) (*livecanary.FileJournal, error) {
+		config, err := readFixtureEndpointConfig(stateDirectory)
+		if err != nil {
+			return nil, err
+		}
+		return livecanary.OpenJournalForPairedFixtureAt(stateDirectory, a, config.AdmissionDirectory)
+	}
+	pairedPrepareJournalForCommand = func(stateDirectory string, a livecanary.Approval) (livecanary.PreparationReceipt, error) {
+		config, err := readFixtureEndpointConfig(stateDirectory)
+		if err != nil {
+			return livecanary.PreparationReceipt{}, err
+		}
+		return livecanary.PreparePairedJournalForFixtureAt(stateDirectory, a, config.AdmissionDirectory)
+	}
+	prepareWorkerJournalForCommand = func(stateDirectory string, a liveworker.Approval) (liveworker.PreparationReceipt, error) {
+		// Worker admission is a distinct disposable fixture root. It is derived
+		// from the worker state identity, never supplied by production approval
+		// or the broker's controller admission root.
+		workerStateReal, err := filepath.EvalSymlinks(stateDirectory)
+		if err != nil || !filepath.IsAbs(workerStateReal) || filepath.Clean(workerStateReal) != workerStateReal {
+			return liveworker.PreparationReceipt{}, liveworker.ErrState
+		}
+		admissionDirectory := filepath.Join(filepath.Dir(workerStateReal), "worker-admission")
+		if err := os.Mkdir(admissionDirectory, 0700); err != nil && !os.IsExist(err) {
+			return liveworker.PreparationReceipt{}, liveworker.ErrState
+		}
+		return livecanary.PrepareWorkerJournalForPairedFixtureAt(stateDirectory, a, admissionDirectory)
+	}
+	newSDKAPIForCommand = func(a livecanary.Approval, c livecanary.Credentials, stateDirectory string) (*livecanary.SDKAPI, error) {
+		config, err := readFixtureEndpointConfig(stateDirectory)
+		if err != nil {
+			return nil, err
+		}
+		return livecanary.NewSDKAPIForPairedFixture(a, c, config.BaseURL, []byte(config.CAPEM))
+	}
+	runPairedTerminalForCommand = func(ctx context.Context, files livecanary.PairedTerminalFiles, c livecanary.Credentials) error {
+		config, err := readFixtureEndpointConfig(files.ControllerStateDirectory)
+		if err != nil {
+			return err
+		}
+		return livecanary.RunPairedTerminalForFixture(ctx, files, c, config.BaseURL, []byte(config.CAPEM), config.AdmissionDirectory)
+	}
+}

experiments/g01-scaleset/cmd/g01-live/main.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/1XP-AI/gh-runnerd/experiments/g01-scaleset/livecanary"
+	"github.com/1XP-AI/gh-runnerd/experiments/g01-scaleset/liveworker"
 )
 
 func buildRevision() (string, bool) {
@@ -40,6 +41,16 @@ func buildRevision() (string, bool) {
 	return version, clean && sdk && len(version) == 40
 }
 
+var openJournalForCommand = livecanary.OpenJournal
+var pairedPrepareJournalForCommand = livecanary.PreparePairedJournal
+var prepareWorkerJournalForCommand = liveworker.PrepareJournal
+var newSDKAPIForCommand = func(a livecanary.Approval, c livecanary.Credentials, _ string) (*livecanary.SDKAPI, error) {
+	return livecanary.NewSDKAPI(a, c)
+}
+var runPairedTerminalForCommand = func(ctx context.Context, files livecanary.PairedTerminalFiles, c livecanary.Credentials) error {
+	return livecanary.RunPairedTerminal(ctx, files, c)
+}
+
 func run(args []string, in io.Reader, out io.Writer) int {
 	return runWithPreparation(args, in, out, buildRevision, livecanary.PrepareJournal)
 }
@@ -56,32 +67,107 @@ func runWithPreparation(args []string, in io.Reader, out io.Writer, revisionForB
 	flags.SetOutput(io.Discard)
 	plan := flags.Bool("plan", false, "")
 	execute := flags.Bool("execute-approved-canary", false, "")
+	pairedExecute := flags.Bool("execute-approved-paired-terminal", false, "")
 	prepare := flags.Bool("prepare-approved-journal", false, "")
+	pairedPrepare := flags.Bool("prepare-approved-paired-journal", false, "")
+	prepareWorker := flags.Bool("prepare-approved-paired-worker-journal", false, "")
 	approvalPath := flags.String("approval", "", "")
 	statePath := flags.String("state-dir", "", "")
 	phase := flags.String("phase", "", "")
+	workerApprovalPath := flags.String("worker-approval", "", "")
+	workerStatePath := flags.String("worker-state-dir", "", "")
+	pairedBinding := flags.String("paired-binding", "", "")
 	reject := func() int {
 		fmt.Fprintln(out, "canary refused; approval, authority or private state requires review")
 		return 1
 	}
 	if flags.Parse(args) != nil || flags.NArg() != 0 {
 		return reject()
 	}
-	if *plan && !*execute && !*prepare {
-		fmt.Fprintln(out, "Controller-only phases: create, before-ack, after-ack, before-acquire, acquire-loss, jit-loss, inspect, cleanup. No worker launch or workflow dispatch. Live execution requires an immutable reviewed build, exact private approval and controller-side broker input.")
+	workerInputs := *workerApprovalPath != "" || *workerStatePath != ""
+	if *plan && !*execute && !*pairedExecute && !*prepare && !*pairedPrepare && !*prepareWorker {
+		if *approvalPath != "" || *statePath != "" || *phase != "" || workerInputs || *pairedBinding != "" {
+			return reject()
+		}
+		fmt.Fprintln(out, "Controller-only phases: create, before-ack, after-ack, before-acquire, acquire-loss, jit-loss, inspect, cleanup. Paired terminal mode uses one same-process executable with explicit worker approval/state inputs and fixed terminal sequencing; no worker launch or workflow dispatch. Live execution requires an immutable reviewed build, exact private approval and controller-side broker input.")
 		return 0
 	}
-	if (*execute == *prepare) || *plan || *approvalPath == "" || *statePath == "" || *phase == "" {
+	modeCount := 0
+	if *execute {
+		modeCount++
+	}
+	if *pairedExecute {
+		modeCount++
+	}
+	if *prepare {
+		modeCount++
+	}
+	if *pairedPrepare {
+		modeCount++
+	}
+	if *prepareWorker {
+		modeCount++
+	}
+	if modeCount != 1 || *plan || *approvalPath == "" || *statePath == "" {
 		return reject()
 	}
+	if *prepareWorker {
+		if *phase != "" || workerInputs || *pairedBinding != "" || prepareWorkerJournalForCommand == nil {
+			return reject()
+		}
+	} else if *pairedExecute {
+		if *phase != "" || *workerApprovalPath == "" || *workerStatePath == "" || *pairedBinding == "" {
+			return reject()
+		}
+	} else if *pairedPrepare {
+		if *phase != "" || workerInputs || *pairedBinding != "" {
+			return reject()
+		}
+	} else if *phase == "" || workerInputs || *pairedBinding != "" {
+		return reject()
+	}
+	if *prepareWorker {
+		worker, workerErr := liveworker.ReadApproval(*approvalPath)
+		if workerErr != nil || worker.Validate(time.Now()) != nil {
+			return reject()
+		}
+		revision, ok := revisionForBuild()
+		if !ok || revision != worker.HarnessSHA {
+			return reject()
+		}
+		receipt, preparationErr := prepareWorkerJournalForCommand(*statePath, worker)
+		if preparationErr != nil || json.NewEncoder(out).Encode(receipt) != nil {
+			return reject()
+		}
+		return 0
+	}
 	a, err := livecanary.ReadApproval(*approvalPath)
-	if err != nil || a.Validate(time.Now()) != nil || !slices.Contains(a.Phases, *phase) {
+	if err != nil || a.Validate(time.Now()) != nil || (!*pairedExecute && !*pairedPrepare && !slices.Contains(a.Phases, *phase)) {
 		return reject()
 	}
 	revision, ok := revisionForBuild()
 	if !ok || revision != a.HarnessSHA {
 		return reject()
 	}
+	var worker liveworker.Approval
+	var binding livecanary.PairedTerminalBinding
+	if *pairedExecute {
+		var workerErr error
+		worker, workerErr = liveworker.ReadApproval(*workerApprovalPath)
+		if workerErr != nil || livecanary.ValidatePairedApprovals(a, worker) != nil || livecanary.ValidatePairedStatePaths(*statePath, *workerStatePath) != nil {
+			return reject()
+		}
+		if livecanary.DecodeStrict([]byte(*pairedBinding), &binding) != nil || livecanary.ValidatePairedTerminalBinding(livecanary.PairedTerminalFiles{ControllerApprovalPath: *approvalPath, ControllerStateDirectory: *statePath, WorkerApprovalPath: *workerApprovalPath, WorkerStateDirectory: *workerStatePath}, binding) != nil {
+			return reject()
+		}
+	}
+	if *pairedPrepare {
+		receipt, e := pairedPrepareJournalForCommand(*statePath, a)
+		if e != nil || json.NewEncoder(out).Encode(receipt) != nil {
+			return reject()
+		}
+		return 0
+	}
 	if *prepare {
 		if prepareJournal == nil {
 			return reject()
@@ -95,11 +181,14 @@ func runWithPreparation(args []string, in io.Reader, out io.Writer, revisionForB
 		}
 		return 0
 	}
-	j, err := livecanary.OpenJournal(*statePath, a)
-	if err != nil {
-		return reject()
+	var j *livecanary.FileJournal
+	if !*pairedExecute {
+		j, err = openJournalForCommand(*statePath, a)
+		if err != nil {
+			return reject()
+		}
+		defer j.Close()
 	}
-	defer j.Close()
 	closable, ok := in.(io.ReadCloser)
 	if !ok {
 		return reject()
@@ -119,7 +208,18 @@ func runWithPreparation(args []string, in io.Reader, out io.Writer, revisionForB
 	if livecanary.DecodeStrict(data, &credentials) != nil {
 		return reject()
 	}
-	api, err := livecanary.NewSDKAPI(a, credentials)
+	if *pairedExecute {
+		if credentials.PairedBinding == nil || *credentials.PairedBinding != binding {
+			return reject()
+		}
+		if runPairedTerminalForCommand(context.Background(), livecanary.PairedTerminalFiles{ControllerApprovalPath: *approvalPath, ControllerStateDirectory: *statePath, WorkerApprovalPath: *workerApprovalPath, WorkerStateDirectory: *workerStatePath}, credentials) != nil {
+			fmt.Fprintln(out, "paired terminal stopped; retain private state and all uncertain resources; no automatic retry")
+			return 1
+		}
+		fmt.Fprintln(out, "paired terminal completed; inspect private evidence")
+		return 0
+	}
+	api, err := newSDKAPIForCommand(a, credentials, *statePath)
 	if err != nil {
 		return reject()
 	}