test: close remaining workspace preprocessing gates

docs/contracts/workspace-preprocessing-cli.md

@@ -26,4 +26,4 @@ The bounded schema-v1 stdin envelope is exact: omitted fields are not equivalent
 
 No other fields, duplicate JSON value, host path, raw SQL, or raw annotation content are accepted. SQL and annotations use a logical basename, base64 bytes, and a declared `sha256:<hex>` digest. The request envelope is independently capped at 24 MiB (after JSON/base64 encoding), enough to carry the frozen 1 MiB-per-file/16 MiB aggregate raw ingress bounds; every supplied field/value must exactly match the command-derived envelope. Candidate responses may carry the internal `hostExport` (`mediaType`, `sha256`, `contentBase64`) only for `suggest-fks`; its object is strict (unknown fields rejected) and is always verified, even without `--output`: YAML media type (`application/yaml` or `text/yaml`), UTF-8, digest, and decoded size at most 700 KiB. It is removed from the public result and written exclusively only after result/run/identity validation. Output publication uses restrictive mode `0600` and refuses existing leaves, links, hardlinks, directories, replacement races, and reparse points.
 
-The public result has schema version 1 and only these fields: `status`, `code`, workspace/revision/descriptor/operation identities, optional run and child run IDs, completed stages, counts, artifact identities, and warnings. Revisions/descriptors are 40-hex; run IDs are 32-hex; artifact digests are `sha256:<hex>`. Allowed statuses are `succeeded`, `unchanged`, `dry_run`, `blocked`, and `failed`. Allowed codes are `ok`, `workspace_not_found`, `workspace_not_activatable`, `binding_missing`, `preprocessing_conflict`, `preprocessing_resume_mismatch`, `manual_review_required`, `evidence_materialization_required`, `effective_config_mismatch`, `semantic_index_incompatible`, `annotation_invalid`, `egress_policy_refused`, and `registry_bootstrap_recovery_conflict`. `succeeded`, `unchanged`, and `dry_run` require `ok` and child exit 0. `blocked` requires one of `manual_review_required`, `evidence_materialization_required`, `preprocessing_conflict`, `preprocessing_resume_mismatch`, or `registry_bootstrap_recovery_conflict`, and child exit 3. `failed` requires a non-`ok` operational code other than those blocked-only codes, and child exit 1. A nonzero child exit is never accepted for another status/code combination. Stdout is capped at 1 MiB and stderr at 64 KiB. Compose is invoked only as `compose run --rm --no-deps --no-TTY workspace-maintenance ...`; output never includes child stderr or secrets.
+The public result has schema version 1 and only these fields: `status`, `code`, workspace/revision/descriptor/operation identities, optional run and child run IDs, completed stages, counts, artifact identities, and warnings. Revisions/descriptors are 40-hex; run IDs are 32-hex; artifact digests are `sha256:<hex>`. Allowed statuses are `succeeded`, `unchanged`, `dry_run`, `blocked`, and `failed`. Allowed codes are `ok`, `workspace_not_found`, `workspace_not_activatable`, `binding_missing`, `preprocessing_conflict`, `preprocessing_resume_mismatch`, `manual_review_required`, `evidence_materialization_required`, `effective_config_mismatch`, `semantic_index_incompatible`, `annotation_invalid`, `egress_policy_refused`, and `registry_bootstrap_recovery_conflict`. `succeeded`, `unchanged`, and `dry_run` require `ok` and child exit 0. `blocked` requires one of `manual_review_required`, `evidence_materialization_required`, `preprocessing_conflict`, `preprocessing_resume_mismatch`, or `registry_bootstrap_recovery_conflict`, and child exit 3. `failed` requires a non-`ok` operational code other than those blocked-only codes, and child exit 1. A nonzero child exit is never accepted for another status/code combination. The public `thothctl` exit mapping is fixed independently of child details: `0` for succeeded/unchanged/dry-run, `3` for an expected blocked result, `2` for command grammar or unsafe host-file failures, and `1` for operational failures (including invalid child envelopes, output-limit failures, and child execution failures). Stdout is capped at 1 MiB after final human/JSON encoding and stderr at 64 KiB after sanitization; output is never allowed to exceed those bounds. In human mode, a `registry_bootstrap_recovery_conflict` result prints exactly `Bootstrap recovery is ambiguous or corrupt; inspect the installation registry jobs.` and prints neither a candidate export nor any run/candidate ID. Compose is invoked only as `compose run --rm --no-deps --no-TTY workspace-maintenance ...`; output never includes child stderr or secrets.

tools/thothctl/cmd/thothctl/main.go

@@ -3,6 +3,7 @@ package main
 
 import (
 	"bufio"
+	"bytes"
 	"context"
 	"encoding/json"
 	"errors"
@@ -22,6 +23,11 @@ import (
 	"github.com/aritmolab/thothii/tools/thothctl/internal/workspaceops"
 )
 
+const (
+	maxPublicStdoutBytes = 1 << 20
+	maxPublicStderrBytes = 64 << 10
+)
+
 const usage = `Usage: thothctl --installation <absolute-path>/thothii-installation.yaml <command>
 
 Commands:
@@ -62,6 +68,10 @@ func main() {
 }
 
 func run(ctx context.Context, args []string, stdout, stderr io.Writer) int {
+	// Apply the public stream bounds after all sanitization and final encoding.
+	// In particular, JSON escaping can expand an otherwise accepted child result.
+	stdout = &boundedWriter{dst: stdout, maximum: maxPublicStdoutBytes}
+	stderr = &boundedWriter{dst: stderr, maximum: maxPublicStderrBytes}
 	if len(args) == 1 && (args[0] == "--help" || args[0] == "-h") {
 		fmt.Fprint(stdout, usage)
 		return 0
@@ -119,7 +129,8 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int {
 			jsonMode = c.JSON
 		}
 		if jsonMode {
-			if err := json.NewEncoder(stdout).Encode(result); err != nil {
+			if err := writeWorkspaceJSON(stdout, result); err != nil {
+				fmt.Fprintln(stderr, "thothctl: workspace result exceeds output limit")
 				return 1
 			}
 		} else {
@@ -221,6 +232,45 @@ func workspaceUsageError(err error) bool {
 		strings.Contains(message, "SQL input exceeds") || strings.Contains(message, "annotation input exceeds")
 }
 
+type boundedWriter struct {
+	dst     io.Writer
+	maximum int64
+	written int64
+}
+
+func (w *boundedWriter) Write(p []byte) (int, error) {
+	remaining := w.maximum - w.written
+	if remaining <= 0 {
+		return 0, io.ErrShortWrite
+	}
+	if int64(len(p)) > remaining {
+		n, err := w.dst.Write(p[:int(remaining)])
+		w.written += int64(n)
+		if err != nil {
+			return n, err
+		}
+		return n, io.ErrShortWrite
+	}
+	n, err := w.dst.Write(p)
+	w.written += int64(n)
+	return n, err
+}
+
+func writeWorkspaceJSON(w io.Writer, result workspaceops.Result) error {
+	var encoded bytes.Buffer
+	encoder := json.NewEncoder(&encoded)
+	// Keep public output compact and avoid HTML-escape amplification of warnings.
+	encoder.SetEscapeHTML(false)
+	if err := encoder.Encode(result); err != nil {
+		return err
+	}
+	if encoded.Len() > maxPublicStdoutBytes {
+		return io.ErrShortWrite
+	}
+	_, err := w.Write(encoded.Bytes())
+	return err
+}
+
 func renderWorkspaceHuman(w io.Writer, result workspaceops.Result) {
 	if result.Code == workspaceops.CodeRegistryBootstrapRecoveryConflict {
 		fmt.Fprintln(w, "Bootstrap recovery is ambiguous or corrupt; inspect the installation registry jobs.")

tools/thothctl/cmd/thothctl/main_test.go

@@ -132,6 +132,42 @@ func TestRunWorkspacePublicDispatchExitMatrix(t *testing.T) {
 	}
 }
 
+func TestRunWorkspaceBoundsFinalJSONEncoding(t *testing.T) {
+	fixture := newCLIFixture(t, "")
+	fixture.setEnvironment(t)
+	result := fmt.Sprintf(`{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"psd","workspaceRevision":"%s","descriptorBlob":"%s","operation":"inspect","completedStages":[],"warnings":[%q]}`, strings.Repeat("0", 40), strings.Repeat("a", 40), strings.Repeat(`"`, 524000))
+	t.Setenv("THOTHCTL_FAKE_WORKSPACE_RESULT", result)
+	var stdout, stderr bytes.Buffer
+	if code := run(context.Background(), []string{"--installation", fixture.installationPath, "workspace", "inspect", "--workspace", "psd", "--json"}, &stdout, &stderr); code != 1 {
+		t.Fatalf("exit = %d, want operational failure 1", code)
+	}
+	if stdout.Len() > 1<<20 {
+		t.Fatalf("stdout length = %d, exceeds 1 MiB", stdout.Len())
+	}
+}
+
+func TestRunWorkspaceBoundsFinalHumanEncoding(t *testing.T) {
+	fixture := newCLIFixture(t, "")
+	fixture.setEnvironment(t)
+	result := fmt.Sprintf(`{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"psd","workspaceRevision":"%s","descriptorBlob":"%s","operation":"inspect","completedStages":[%q]}`, strings.Repeat("0", 40), strings.Repeat("a", 40), strings.Repeat("x", (1<<20)-100))
+	t.Setenv("THOTHCTL_FAKE_WORKSPACE_RESULT", result)
+	var stdout, stderr bytes.Buffer
+	_ = run(context.Background(), []string{"--installation", fixture.installationPath, "workspace", "inspect", "--workspace", "psd"}, &stdout, &stderr)
+	if stdout.Len() > 1<<20 {
+		t.Fatalf("stdout length = %d, exceeds 1 MiB", stdout.Len())
+	}
+}
+
+func TestRunBoundsParseErrorStderr(t *testing.T) {
+	fixture := newCLIFixture(t, "")
+	fixture.setEnvironment(t)
+	var stdout, stderr bytes.Buffer
+	_ = run(context.Background(), []string{"--installation", fixture.installationPath, "workspace", "inspect", "--workspace", "psd", "--unknown", strings.Repeat("x", 70<<10)}, &stdout, &stderr)
+	if stderr.Len() > 64<<10 {
+		t.Fatalf("stderr length = %d, exceeds 64 KiB", stderr.Len())
+	}
+}
+
 func TestRunWorkspaceOperationalFailureExitsOne(t *testing.T) {
 	fixture := newCLIFixture(t, "")
 	fixture.setEnvironment(t)

tools/thothctl/internal/compose/process_windows_test.go

@@ -3,9 +3,15 @@
 package compose
 
 import (
-	"golang.org/x/sys/windows"
+	"context"
+	"errors"
+	"os"
 	"os/exec"
+	"path/filepath"
 	"testing"
+	"time"
+
+	"golang.org/x/sys/windows"
 )
 
 func TestOwnedProcessUsesCreationTimeJobAttribute(t *testing.T) {
@@ -24,3 +30,74 @@ func TestOwnedJobPolicyKillsDescendantsOnClose(t *testing.T) {
 		t.Fatal("owned job must use kill-on-close policy")
 	}
 }
+
+// TestWindowsProcessHelper is both a test helper and the child process used by the runtime
+// ownership regression below. The child writes a marker only if it outlives its parent job.
+func TestWindowsProcessHelper(t *testing.T) {
+	mode := os.Getenv("THOTHCTL_WINDOWS_HELPER")
+	if mode == "" {
+		return
+	}
+	marker := os.Getenv("THOTHCTL_WINDOWS_MARKER")
+	if mode == "child" {
+		time.Sleep(2 * time.Second)
+		_ = os.WriteFile(marker, []byte("survived"), 0o600)
+		return
+	}
+	child := exec.Command(os.Args[0], "-test.run=TestWindowsProcessHelper", "--")
+	child.Env = append(os.Environ(), "THOTHCTL_WINDOWS_HELPER=child", "THOTHCTL_WINDOWS_MARKER="+marker)
+	if err := child.Start(); err != nil {
+		t.Fatal(err)
+	}
+	if mode == "exit" {
+		return
+	}
+	if mode == "cancel" {
+		for {
+			time.Sleep(time.Second)
+		}
+	}
+	for {
+		_, _ = os.Stdout.Write([]byte("overflow\n"))
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func TestOwnedProcessTerminatesDescendantOnOverflowCancelAndParentExit(t *testing.T) {
+	for _, mode := range []string{"overflow", "cancel", "exit"} {
+		t.Run(mode, func(t *testing.T) {
+			marker := filepath.Join(t.TempDir(), "descendant-marker")
+			t.Setenv("THOTHCTL_WINDOWS_HELPER", mode)
+			t.Setenv("THOTHCTL_WINDOWS_MARKER", marker)
+			ctx := context.Background()
+			var cancel context.CancelFunc
+			if mode == "cancel" {
+				ctx, cancel = context.WithCancel(ctx)
+			}
+			result := make(chan error, 1)
+			go func() {
+				_, err := runBoundedWindows(ctx, os.Args[0], []string{"-test.run=TestWindowsProcessHelper", "--"}, nil, CaptureLimits{StdoutBytes: 1024, StderrBytes: 1024})
+				result <- err
+			}()
+			if mode == "cancel" {
+				time.Sleep(100 * time.Millisecond)
+				cancel()
+			}
+			select {
+			case err := <-result:
+				if mode == "overflow" && !errors.Is(err, ErrOutputLimit) {
+					t.Fatalf("overflow error = %v, want ErrOutputLimit", err)
+				}
+				if mode == "cancel" && !errors.Is(err, context.Canceled) {
+					t.Fatalf("cancel error = %v, want context.Canceled", err)
+				}
+			case <-time.After(5 * time.Second):
+				t.Fatal("owned process did not terminate")
+			}
+			time.Sleep(150 * time.Millisecond)
+			if _, err := os.Stat(marker); !errors.Is(err, os.ErrNotExist) {
+				t.Fatalf("descendant marker error = %v; descendant survived %s termination", err, mode)
+			}
+		})
+	}
+}

tools/thothctl/internal/safeio/files_test.go

@@ -58,6 +58,16 @@ func TestReadCanonicalUTF8RejectsNonUTF8AndBounds(t *testing.T) {
 	}
 }
 
+func TestReadCanonicalUTF8RejectsSQLLargerThanOneMiB(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "schema.sql")
+	if err := os.WriteFile(path, make([]byte, (1<<20)+1), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ReadCanonicalUTF8(path, 1<<20); !errors.Is(err, ErrUnsafeFile) {
+		t.Fatalf("ReadCanonicalUTF8 1 MiB + 1 SQL = %v, want ErrUnsafeFile", err)
+	}
+}
+
 func TestWriteCanonicalExclusiveRejectsExistingAndCreatesPrivateFile(t *testing.T) {
 	root, err := filepath.EvalSymlinks(t.TempDir())
 	if err != nil {

tools/thothctl/internal/safeio/files_unix_test.go

@@ -3,10 +3,12 @@
 package safeio
 
 import (
+	"bytes"
 	"errors"
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"golang.org/x/sys/unix"
 )
@@ -31,6 +33,56 @@ func TestReadCanonicalRegularRejectsNamedPipeWithoutBlocking(t *testing.T) {
 	}
 }
 
+func TestReadCanonicalRegularRejectsReplacementDuringRead(t *testing.T) {
+	root := t.TempDir()
+	path := filepath.Join(root, "schema.sql")
+	replacement := filepath.Join(root, "replacement.sql")
+	parked := filepath.Join(root, "parked.sql")
+	contents := bytes.Repeat([]byte("x"), 64<<20)
+	if err := os.WriteFile(path, contents, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(replacement, contents, 0o600); err != nil {
+		t.Fatal(err)
+	}
+
+	var caught bool
+	for attempt := 0; attempt < 3 && !caught; attempt++ {
+		result := make(chan error, 1)
+		go func() {
+			_, err := ReadCanonicalRegular(path, int64(len(contents)))
+			result <- err
+		}()
+		time.Sleep(time.Millisecond)
+		var finalErr error
+		for i := 0; i < 20; i++ {
+			if err := os.Rename(path, parked); err == nil {
+				if err := os.Rename(replacement, path); err != nil {
+					t.Fatal(err)
+				}
+				time.Sleep(time.Millisecond)
+				if err := os.Rename(parked, replacement); err != nil {
+					t.Fatal(err)
+				}
+			}
+			select {
+			case finalErr = <-result:
+				i = 20
+			default:
+			}
+		}
+		if finalErr == nil {
+			finalErr = <-result
+		}
+		if errors.Is(finalErr, ErrUnsafeFile) {
+			caught = true
+		}
+	}
+	if !caught {
+		t.Fatal("replacement during read was not rejected")
+	}
+}
+
 func TestCanonicalDescriptorOwnershipDoesNotLeakAcrossNestedOperations(t *testing.T) {
 	root, err := filepath.EvalSymlinks(t.TempDir())
 	if err != nil {

tools/thothctl/internal/workspaceops/operations_test.go

@@ -299,10 +299,28 @@ func TestParseWorkspaceAcceptsAssumptionAndEnforcesLimits(t *testing.T) {
 	if _, err := ParseWorkspaceCommand([]string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", "orders.id=customers.id"}); err != nil {
 		t.Fatal(err)
 	}
-	if _, err := ParseWorkspaceCommand([]string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", strings.Repeat("a", maxAssumptionBytes+1)}); err == nil {
+	exactAssumption := strings.Repeat("a", 127) + "=" + strings.Repeat("b", 128)
+	if len(exactAssumption) != maxAssumptionBytes {
+		t.Fatalf("test assumption length = %d, want %d", len(exactAssumption), maxAssumptionBytes)
+	}
+	if _, err := ParseWorkspaceCommand([]string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", exactAssumption}); err != nil {
+		t.Fatalf("rejected exact assumption bound: %v", err)
+	}
+	if _, err := ParseWorkspaceCommand([]string{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--assume", exactAssumption + "a"}); err == nil {
 		t.Fatal("accepted oversized assumption")
 	}
 	args := []string{"workspace", "schema", "suggest-fks", "--workspace", "psd"}
+	for i := 0; i < maxAssumptions; i++ {
+		args = append(args, "--assume", fmt.Sprintf("a%d=b%d", i, i))
+	}
+	if _, err := ParseWorkspaceCommand(args); err != nil {
+		t.Fatalf("rejected exact assumption count: %v", err)
+	}
+	args = append(args, "--assume", "overflow=overflow")
+	if _, err := ParseWorkspaceCommand(args); err == nil {
+		t.Fatalf("accepted %d assumptions", maxAssumptions+1)
+	}
+	args = []string{"workspace", "schema", "suggest-fks", "--workspace", "psd"}
 	for i := 0; i < 33; i++ {
 		args = append(args, "--from-sql", fmt.Sprintf("schema-%d.sql", i))
 	}
@@ -368,7 +386,21 @@ func TestWorkspaceIngressAcceptsExactCountsAndByteBounds(t *testing.T) {
 	for i := 0; i < maxAssumptions; i++ {
 		assumptions = append(assumptions, fmt.Sprintf("column_%d=table_%d", i, i))
 	}
-	command := SuggestFksRequest{WorkspaceID: "psd", FromSQL: paths, Assume: assumptions}
+	argv := []string{"workspace", "schema", "suggest-fks", "--workspace", "psd"}
+	for _, path := range paths {
+		argv = append(argv, "--from-sql", path)
+	}
+	for _, assumption := range assumptions {
+		argv = append(argv, "--assume", assumption)
+	}
+	parsed, err := ParseWorkspaceCommand(argv)
+	if err != nil {
+		t.Fatalf("parser rejected exact file/assumption counts: %v", err)
+	}
+	command, ok := parsed.(SuggestFksRequest)
+	if !ok {
+		t.Fatalf("parsed command = %T, want SuggestFksRequest", parsed)
+	}
 	for _, path := range paths {
 		if b, e := safeio.ReadCanonicalUTF8(path, maxSQLFile); e != nil {
 			t.Fatalf("direct read %s: %v", path, e)