Fix bounded Rust analysis and prepare 0.29.0-beta.3

.code-polishy.json

@@ -332,6 +332,7 @@
           "internal/engine/report_types.go",
           "internal/engine/format.go",
           "internal/engine/gate_command_runner.go",
+          "internal/engine/gate_stream_test.go",
           "internal/engine/gate_run.go",
           "internal/engine/test_evidence.go",
           "internal/gaterun/run.go",
@@ -449,7 +450,7 @@
     {
       "name": "rustfacts",
       "paths": ["internal/rustfacts/**"],
-      "dependsOn": ["policy", "repository", "runner", "release"]
+      "dependsOn": ["policy", "pythonfacts", "repository", "runner", "release"]
     },
     {
       "name": "runner",

CHANGELOG.md

@@ -2,6 +2,17 @@
 
 ## Unreleased
 
+## 0.29.0-beta.3 - 2026-10-17
+
+- Resolve configured prerequisites through the installed tool inventory, including
+  the optional Cargo companion, without relying on ambient PATH entries.
+- Stream Rust syntax facts as bounded source records so expanded results larger
+  than 8 MiB retain complete coverage. Oversized individual results remain
+  explicit coverage failures while valid sibling sources remain available.
+- Route Rust syntax workers through the shared Python program transport and
+  preserve their planned command identities, bounded logs, and failed attempts
+  throughout gate execution.
+
 ## 0.29.0-beta.2 - 2026-10-16
 
 - Recognize pinned OSV-Scanner, ty and Python tools on Windows, where files

VERSION

@@ -1 +1 @@
-0.29.0-beta.2
+0.29.0-beta.3

docs/design/cargo-facts.md

@@ -108,6 +108,17 @@ oversized source or a failed batch produces coverage findings for the affected
 paths without discarding valid siblings. Cancellation prevents later batches
 from starting and leaves their files explicitly uncovered.
 
+Responses stream one newline-terminated `rust-source/v2` record per requested
+source, in request order, through the common runner's writer interface. Binary
+UTF-8 output uses a single LF terminator on every host, including Windows. Each
+record is limited to 8388608 bytes including its terminator and validated before
+the next is retained; diagnostics remain limited to 65536 bytes. Syntax facts
+can expand beyond their input size without creating a repository-wide response
+ceiling. An individual source whose facts exceed the record budget emits an
+explicit incomplete result, preserving its valid siblings. A missing, repeated,
+foreign, malformed, oversized or unterminated record, or a failed worker,
+invalidates its batch. No partial stream establishes complete coverage.
+
 Locations unpack Tree-sitter's point tuples and translate byte columns into
 original Unicode coordinates. The pinned Python binding's
 [point attribute getters](https://github.com/tree-sitter/py-tree-sitter/blob/v0.26.0/tree_sitter/binding/point.c)
@@ -309,3 +320,9 @@ probe before atomic replacement. Corrupt downloads, invalid payloads, failed
 probes and replacement failures preserve a previous working executable. Reuse
 checks the installed release; it does not hash local state. This provisions a
 runner, not test evidence or doctest execution.
+
+The syntax worker uses the shared Python program transport: a bounded program
+record precedes the request on standard input. Command arguments contain only
+the fixed bootstrap, so multiline source remains valid in gate identities and
+does not depend on platform command-line limits. Rust facts depend on the
+existing Python transport owner rather than maintaining a second bootstrap.

docs/design/execution-observability.md

@@ -28,6 +28,16 @@ identifies both scope and cost.
 The observation wrapper forwards structured process results unchanged to suite
 execution. Recording an exit status while discarding it from the caller would
 make native test-artifact validation disagree with the command that produced it.
+Configured prerequisite checks resolve registered executables through the
+governed tool inventory. Doctor uses these absolute paths instead of ambient
+PATH entries, so an ambient executable cannot conceal a missing member of an
+installed tool bundle. Unregistered external commands retain ordinary host
+discovery.
+Gate execution forwards typed streams through the same planned-command, log,
+and attempt-recording path. Consumers validate bounded records incrementally;
+the diagnostic log retains its existing bounded head and tail. Streaming does
+not introduce an aggregate capture limit or permit reuse without the facts.
+Consumer failures remain failed command attempts.
 Bounded capture overflow is an operational failure even when the child exits with
 a normal test-failure code. A partial stream cannot establish native outcome
 evidence or a regression-proof red result. Cancellation and timeout retain their

docs/design/historical-review-policy.md

@@ -19,6 +19,9 @@ historical ownership relations explicitly for impact analysis, without
 inventing a focused suite or changing the candidate's ownership rules. Suite
 defaults preserve the command identity used by historical review evidence.
 Only the candidate configuration governs current checks and test execution.
+The planned runner enforces the same command identity for captured and streamed
+analyzers; stream forwarding retains the gate attempt and log boundary described
+in [Execution observability](execution-observability.md).
 
 The engine reads the snapshot from the bounded regular Git blob at the exact
 base revision. An absent configuration selects first-adoption behavior;

internal/architecture/rust_test.go

@@ -2,6 +2,7 @@ package architecture
 
 import (
 	"context"
+	"io"
 	"path/filepath"
 	"slices"
 	"strings"
@@ -298,3 +299,8 @@ func (boundary *rustGraphRecordingRunner) RunStructured(ctx context.Context, roo
 	boundary.names = append(boundary.names, command.Name)
 	return boundary.OSRunner.RunStructured(ctx, root, command)
 }
+
+func (boundary *rustGraphRecordingRunner) RunWithWriters(ctx context.Context, root string, command policy.Command, stdout, stderr io.Writer) (runner.Result, error) {
+	boundary.names = append(boundary.names, command.Name)
+	return boundary.OSRunner.RunWithWriters(ctx, root, command, stdout, stderr)
+}

internal/engine/engine.go

@@ -1021,8 +1021,10 @@ func checkedInCommandFinding(repo repository.Repository, name, executable string
 }
 
 func configuredToolPath(repo repository.Repository, executable string) string {
-	if executable == "go" || executable == "gofmt" {
-		return repo.GoTool(executable)
+	for _, tool := range repo.CommandEnvironment().Tools {
+		if tool.Name == executable {
+			return tool.Path
+		}
 	}
 	return executable
 }

internal/engine/gate_command_runner.go

@@ -106,20 +106,49 @@ func (commandRunner *gateArtifactRunner) Run(ctx context.Context, root string, c
 }
 
 func (commandRunner *gateArtifactRunner) RunWithResult(ctx context.Context, root string, command policy.Command) (runner.Result, error) {
-	result, _, err := commandRunner.runNext(ctx, root, command, false, nil)
+	result, _, err := commandRunner.runNext(ctx, root, command, gateCommandOutput{})
 	return result, err
 }
 
 func (commandRunner *gateArtifactRunner) RunWithOutput(ctx context.Context, root string, command policy.Command) (runner.Result, runner.Output, error) {
-	return commandRunner.runNext(ctx, root, command, true, nil)
+	return commandRunner.runNext(ctx, root, command, gateCommandOutput{capture: true})
 }
 
 func (commandRunner *gateArtifactRunner) RunWithReportOutput(ctx context.Context, root string, command policy.Command, accept runner.ReportOutcomeAcceptor) (runner.Result, runner.Output, error) {
-	return commandRunner.runNext(ctx, root, command, true, accept)
+	return commandRunner.runNext(ctx, root, command, gateCommandOutput{capture: true, accept: accept})
 }
 
 func (commandRunner *gateArtifactRunner) RunStructured(ctx context.Context, root string, command policy.Command) (runner.Result, runner.Output, error) {
-	return commandRunner.runNext(ctx, root, command, true, nil)
+	return commandRunner.runNext(ctx, root, command, gateCommandOutput{capture: true})
+}
+
+func (commandRunner *gateArtifactRunner) RunWithWriters(ctx context.Context, root string, command policy.Command, stdout, stderr io.Writer) (runner.Result, error) {
+	if stdout == nil {
+		stdout = io.Discard
+	}
+	if stderr == nil {
+		stderr = io.Discard
+	}
+	result, _, err := commandRunner.runNext(ctx, root, command, gateCommandOutput{stdout: stdout, stderr: stderr})
+	return result, err
+}
+
+type gateCommandOutput struct {
+	capture bool
+	accept  runner.ReportOutcomeAcceptor
+	stdout  io.Writer
+	stderr  io.Writer
+}
+
+func (output gateCommandOutput) run(ctx context.Context, delegate runner.Runner, root string, command policy.Command, stdout, stderr io.Writer) (runner.Result, runner.Output, error) {
+	if output.stdout != nil {
+		if _, ok := delegate.(runner.StreamRunner); !ok {
+			return runner.Result{ExitStatus: -1, FailureCategory: runner.FailureOperational}, runner.Output{}, fmt.Errorf("gate command %q requires streamed output", command.Name)
+		}
+		stdout = io.MultiWriter(stdout, output.stdout)
+		stderr = io.MultiWriter(stderr, output.stderr)
+	}
+	return runGateCommand(ctx, delegate, root, command, stdout, stderr, output.capture || testartifact.NeedsCommandOutput(command))
 }
 
 func (commandRunner *gateArtifactRunner) ManagesTestArtifacts() bool { return true }
@@ -188,7 +217,7 @@ func (commandRunner *gateArtifactRunner) TestRetryRunner() runner.Runner {
 	return &gateRetryRunner{parent: commandRunner}
 }
 
-func (commandRunner *gateArtifactRunner) runNext(ctx context.Context, root string, command policy.Command, captureOutput bool, accept runner.ReportOutcomeAcceptor) (runner.Result, runner.Output, error) {
+func (commandRunner *gateArtifactRunner) runNext(ctx context.Context, root string, command policy.Command, output gateCommandOutput) (runner.Result, runner.Output, error) {
 	if commandRunner.err != nil {
 		return runner.Result{ExitStatus: -1, FailureCategory: runner.FailureOperational}, runner.Output{}, commandRunner.err
 	}
@@ -206,7 +235,7 @@ func (commandRunner *gateArtifactRunner) runNext(ctx context.Context, root strin
 		fmt.Fprintf(commandRunner.progress, "REUSE %s\n", command.Name)
 		return runner.Result{ExitStatus: 0}, runner.Output{}, nil
 	}
-	return commandRunner.execute(ctx, root, index, command, captureOutput, false, accept)
+	return commandRunner.execute(ctx, root, index, command, output, false)
 }
 
 func (commandRunner *gateArtifactRunner) plannedIndex(root string, command policy.Command) (int, error) {
@@ -233,7 +262,7 @@ func (commandRunner *gateArtifactRunner) plannedIndex(root string, command polic
 	return index, nil
 }
 
-func (commandRunner *gateArtifactRunner) execute(ctx context.Context, root string, index int, command policy.Command, captureOutput, diagnostic bool, accept runner.ReportOutcomeAcceptor) (runner.Result, runner.Output, error) {
+func (commandRunner *gateArtifactRunner) execute(ctx context.Context, root string, index int, command policy.Command, options gateCommandOutput, diagnostic bool) (runner.Result, runner.Output, error) {
 	fmt.Fprintf(commandRunner.progress, "RUN %s\n", command.Name)
 	actual, err := commandRunner.prepareCommand(index, command)
 	if err != nil {
@@ -245,7 +274,7 @@ func (commandRunner *gateArtifactRunner) execute(ctx context.Context, root strin
 		commandRunner.err = err
 		return runner.Result{ExitStatus: -1, FailureCategory: runner.FailureOperational}, runner.Output{}, err
 	}
-	result, output, runErr := runGateCommand(ctx, commandRunner.delegate, root, actual, commandLog.Stdout(), commandLog.Stderr(), captureOutput || testartifact.NeedsCommandOutput(actual))
+	result, output, runErr := options.run(ctx, commandRunner.delegate, root, actual, commandLog.Stdout(), commandLog.Stderr())
 	runErr = captureGateTestOutput(ctx, actual, &result, output, runErr)
 	artifacts, artifactErr := commandRunner.validateTestArtifacts(index, actual, &result)
 	runErr = errors.Join(runErr, artifactErr)
@@ -254,7 +283,7 @@ func (commandRunner *gateArtifactRunner) execute(ctx context.Context, root strin
 		commandRunner.err = logErr
 		return runner.Result{ExitStatus: -1, FailureCategory: runner.FailureOperational}, output, errors.Join(runErr, logErr)
 	}
-	status, category, reportBearing := gateCommandAttempt(ctx, result, output, runErr, accept)
+	status, category, reportBearing := gateCommandAttempt(ctx, result, output, runErr, options.accept)
 	gateCategory := gaterun.FailureCategory("")
 	if status == gaterun.Failed {
 		gateCategory = gaterun.FailureCategory(category)
@@ -345,7 +374,7 @@ func (commandRunner *gateArtifactRunner) runDiagnostic(ctx context.Context, root
 		commandRunner.err = errors.Join(commandRunner.err, err)
 		return runner.Result{ExitStatus: -1, FailureCategory: runner.FailureOperational}, err
 	}
-	result, _, err := commandRunner.execute(ctx, root, index, command, false, true, nil)
+	result, _, err := commandRunner.execute(ctx, root, index, command, gateCommandOutput{}, true)
 	return result, err
 }
 
@@ -357,7 +386,7 @@ func (commandRunner *gateArtifactRunner) runRetry(ctx context.Context, root stri
 		commandRunner.err = errors.Join(commandRunner.err, err)
 		return runner.Result{ExitStatus: -1, FailureCategory: runner.FailureOperational}, err
 	}
-	result, _, err := commandRunner.execute(ctx, root, index, command, false, false, nil)
+	result, _, err := commandRunner.execute(ctx, root, index, command, gateCommandOutput{}, false)
 	return result, err
 }
 
@@ -488,7 +517,7 @@ func (commandRunner *gateArtifactRunner) runTestPrerequisite(ctx context.Context
 	if _, failed := commandRunner.failedTests[suite]; failed {
 		for index, expected := range commandRunner.expected {
 			if expected.Category == gaterun.Check && samePolicyCommand(expected.Command, command) {
-				return commandRunner.execute(ctx, root, index, command, true, false, nil)
+				return commandRunner.execute(ctx, root, index, command, gateCommandOutput{capture: true}, false)
 			}
 		}
 	}

internal/engine/gate_stream_test.go

@@ -0,0 +1,134 @@
+package engine
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/riteofstring/code-polishy/internal/gaterun"
+	"github.com/riteofstring/code-polishy/internal/repository"
+	"github.com/riteofstring/code-polishy/internal/runner"
+	"github.com/riteofstring/code-polishy/internal/rustfacts"
+)
+
+func TestGateStreamsLargeRustFactsAndRecordsBoundedLogs(t *testing.T) {
+	t.Parallel()
+	repo := repository.Repository{Root: t.TempDir(), PolicyRoot: enginePolicyRoot(t)}
+	var source strings.Builder
+	for index := range 1200 {
+		fmt.Fprintf(&source, "pub fn function_%d() {}\n", index)
+	}
+	paths := make([]string, 20)
+	for index := range paths {
+		paths[index] = fmt.Sprintf("source_%02d.rs", index)
+		writeEngineFile(t, repo.Root, paths[index], source.String(), 0o600)
+	}
+	plan := rustfacts.PlanSources(repo, paths)
+	boundary := rustSourceGate(t, repo, plan)
+	planned := &mergeGatePlannedRunner{root: repo.Root, delegate: boundary, expected: boundary.expected}
+	result := rustfacts.ReadSources(t.Context(), repo, plan, planned)
+	if len(result.Findings) != 0 || len(result.Files) != len(paths) || planned.next != 1 {
+		t.Fatalf("gate lost streamed syntax evidence: files=%d findings=%+v", len(result.Files), result.Findings)
+	}
+	for index, file := range result.Files {
+		if file.Path != paths[index] || len(file.Functions) != 1200 || file.Functions[1199].Name != "function_1199" {
+			t.Fatalf("gate changed parsed function inventory for %s", file.Path)
+		}
+	}
+	outcome := finishRustSourceGate(t, boundary, gaterun.RunPassed)
+	data, err := os.ReadFile(filepath.Join(repo.Root, outcome.Attempts[0].LogPath))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var log struct {
+		StdoutBytes     int64  `json:"stdout_bytes"`
+		StdoutTruncated bool   `json:"stdout_truncated"`
+		StdoutTail      []byte `json:"stdout_tail"`
+	}
+	if err := json.Unmarshal(data, &log); err != nil {
+		t.Fatal(err)
+	}
+	if log.StdoutBytes <= 8<<20 || !log.StdoutTruncated || !strings.Contains(string(log.StdoutTail), "function_1199") {
+		t.Fatalf("gate lost its bounded large-stream log: bytes=%d truncated=%v", log.StdoutBytes, log.StdoutTruncated)
+	}
+}
+
+func TestGateRecordsStreamConsumerFailures(t *testing.T) {
+	t.Parallel()
+	repo := repository.Repository{Root: t.TempDir(), PolicyRoot: enginePolicyRoot(t)}
+	writeEngineFile(t, repo.Root, "source.rs", "pub fn example() {}\n", 0o600)
+	plan := rustfacts.PlanSources(repo, []string{"source.rs"})
+	boundary := rustSourceGate(t, repo, plan)
+	planned := &mergeGatePlannedRunner{root: repo.Root, delegate: boundary, expected: boundary.expected}
+	result, err := planned.RunWithWriters(t.Context(), repo.Root, plan.Commands[0], rejectingGateStream{}, io.Discard)
+	if err == nil || result.FailureCategory != runner.FailureOperational {
+		t.Fatalf("consumer failure was accepted: result=%+v error=%v", result, err)
+	}
+	outcome := finishRustSourceGate(t, boundary, gaterun.RunFailed)
+	if outcome.Attempts[0].FailureCategory != gaterun.FailureCategory(runner.FailureOperational) {
+		t.Fatalf("gate lost the stream failure category: %+v", outcome)
+	}
+}
+
+func TestGateRejectsUnplannedRustStream(t *testing.T) {
+	t.Parallel()
+	repo := repository.Repository{Root: t.TempDir(), PolicyRoot: enginePolicyRoot(t)}
+	writeEngineFile(t, repo.Root, "source.rs", "pub fn example() {}\n", 0o600)
+	plan := rustfacts.PlanSources(repo, []string{"source.rs"})
+	boundary := rustSourceGate(t, repo, plan)
+	planned := &mergeGatePlannedRunner{root: repo.Root, delegate: boundary, expected: boundary.expected}
+	command := plan.Commands[0]
+	command.Stdin = append(append([]byte{}, command.Stdin...), '\n')
+	result, err := planned.RunWithWriters(t.Context(), repo.Root, command, io.Discard, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "request") || result.FailureCategory != runner.FailureOperational || boundary.next != 0 {
+		t.Fatalf("unplanned stream reached gate execution: result=%+v next=%d error=%v", result, boundary.next, err)
+	}
+}
+
+type rejectingGateStream struct{}
+
+func (rejectingGateStream) Write([]byte) (int, error) {
+	return 0, errors.New("syntax consumer rejected the output")
+}
+
+func rustSourceGate(t *testing.T, repo repository.Repository, plan rustfacts.SourcePlan) *gateArtifactRunner {
+	t.Helper()
+	if len(plan.Findings) != 0 || len(plan.Commands) != 1 {
+		t.Fatalf("unexpected parser plan: %+v", plan.Findings)
+	}
+	planned := MergeGateExecutionCommand{Root: repo.Root, Category: gaterun.Check, Scope: "repository", Cost: "quick", Command: plan.Commands[0]}
+	identity, err := gaterun.NewIdentity(gaterun.IdentityInput{
+		Gate: gaterun.MergeGate, RequestedBase: "main", ExactBase: strings.Repeat("a", 40), Candidate: strings.Repeat("b", 40), PolicyLevel: "recommended",
+		Release: gaterun.ReleaseIdentity{Version: "test", Digest: strings.Repeat("a", 64)}, ConfigurationSHA256: strings.Repeat("b", 64),
+		Platform: gaterun.Platform{OS: "test", Arch: "test"}, Commands: []gaterun.CommandSpec{gateRunCommandSpec(planned, nil)},
+		Environment: []gaterun.EnvironmentInput{}, AmbientEnvironment: []gaterun.EnvironmentInput{}, BehaviorReview: testGateRunBehaviorReview(),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	run, err := gaterun.Start(gaterun.StartOptions{RepositoryRoot: repo.Root, Identity: identity})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return &gateArtifactRunner{delegate: runner.OSRunner{}, run: run, expected: []MergeGateExecutionCommand{planned}, progress: io.Discard}
+}
+
+func finishRustSourceGate(t *testing.T, boundary *gateArtifactRunner, status gaterun.RunStatus) gaterun.CommandOutcome {
+	t.Helper()
+	report, err := boundary.run.Finalize(gaterun.FinalizeOptions{Status: status, BehaviorReview: testGateRunBehaviorReview()})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := gaterun.LoadReport(boundary.expected[0].Root, report.Identity); err != nil {
+		t.Fatal(err)
+	}
+	if len(report.Commands) != 1 || len(report.Commands[0].Attempts) != 1 || string(report.Commands[0].Status) != string(status) {
+		t.Fatalf("gate did not record its command attempt: %+v", report.Commands)
+	}
+	return report.Commands[0]
+}