fix: harden workspace preprocessing contract

docs/contracts/workspace-preprocessing-cli.md

@@ -14,8 +14,16 @@ workspace preprocess run --workspace ID [--resume RUN] [--json]
 
 IDs are lowercase `[a-z][a-z0-9-]{2,62}` and outer run IDs are exactly 32 lowercase hexadecimal characters. `--from-sql` accepts at most 32 canonical regular non-symlink files, each at most 1 MiB and 16 MiB total. `--assume` accepts at most 256 values of 256 bytes, each matching `column=table`. Annotations are one canonical UTF-8 file at most 16 MiB. Existing output, links, hardlinks, directories, replacement races, and reparse points are refused with a generic unsafe-file error.
 
-The bounded schema-v1 stdin request contains only operation-appropriate fields. SQL and annotations are represented by a logical basename, base64 bytes, and a declared `sha256:<hex>` digest; no host path or raw content enters the request. The complete request is at most 1 MiB, has exactly one JSON value, and cannot replace command-derived fields. Candidate responses may carry the internal `hostExport` (`mediaType`, `sha256`, `contentBase64`) only for `suggest-fks`; it is verified (UTF-8 YAML, digest, and 700 KiB maximum), removed from the public result, and written exclusively only after result/run/identity validation.
+The bounded schema-v1 stdin envelope is exact: omitted fields are not equivalent to explicit zero, empty, or `null` fields. Its fields are, per operation (in addition to the always-required `schemaVersion:1`, `operation`, and `workspaceId`):
 
-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>`. A nonzero child exit is accepted only for the matching blocked (3) or failed (1) result. 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.
+* `inspect`: no additional fields.
+* `dwh`: `resume` only when `--resume RUN` was supplied.
+* `run`: `resume` only when `--resume RUN` was supplied.
+* `evidence`: `resume` only when supplied, and `dryRun` only when `--dry-run` was supplied.
+* `suggest-fks`: `sql` only when one or more `--from-sql` files were supplied, and `assume` only when one or more `--assume` values were supplied. Each `sql` item is exactly `{basename, contentBase64, sha256}`.
+* `check`: `resume` is required; `annotations` and `reviewedCandidates` are either both present or both absent. `annotations` is exactly `{basename, contentBase64, sha256}`.
+* `index-schema`: no additional fields.
 
-Exit 0 means succeeded, unchanged, or dry-run; exit 3 means an expected operator checkpoint/block; exit 2 means grammar or unsafe local-file failure; exit 1 means operational failure. Human mode prints only allowlisted identity/status fields. For `registry_bootstrap_recovery_conflict` it prints exactly: `Bootstrap recovery is ambiguous or corrupt; inspect the installation registry jobs.`
+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 complete request is at most 1 MiB and 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.

tools/thothctl/cmd/thothctl/main.go

@@ -95,7 +95,7 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int {
 		}
 		result, operationErr := workspaceops.Run(ctx, installation, runner, workspaceCommand, nil)
 		if operationErr != nil {
-			if errors.Is(operationErr, compose.ErrOutputLimit) || strings.Contains(operationErr.Error(), "unsafe") || strings.Contains(operationErr.Error(), "exceeds") || strings.Contains(operationErr.Error(), "invalid workspace") {
+			if workspaceUsageError(operationErr) {
 				return commandUsageError(stderr, operationErr.Error())
 			}
 			fmt.Fprintf(stderr, "thothctl: %s\n", output.Sanitize(operationErr.Error(), secretValues))
@@ -210,6 +210,17 @@ func run(ctx context.Context, args []string, stdout, stderr io.Writer) int {
 	return writeResult(result, err, secretValues, stdout, stderr)
 }
 
+func workspaceUsageError(err error) bool {
+	if err == nil || errors.Is(err, compose.ErrOutputLimit) {
+		return false
+	}
+	message := err.Error()
+	// These are host-side grammar/local-file failures. Child envelope/result and
+	// bounded-stream failures are operational and deliberately remain exit 1.
+	return strings.Contains(message, "unsafe") || strings.Contains(message, "request exceeds") ||
+		strings.Contains(message, "SQL input exceeds") || strings.Contains(message, "annotation input exceeds")
+}
+
 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

@@ -3,6 +3,7 @@ package main
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -104,6 +105,17 @@ func TestUsageDocumentsClosedConfigureUpdateSourcesAndMaintenanceRecovery(t *tes
 	}
 }
 
+func TestRunWorkspaceRejectsInvalidCommandBeforeDocker(t *testing.T) {
+	fixture := newCLIFixture(t, "")
+	fixture.setEnvironment(t)
+	var stdout, stderr bytes.Buffer
+	code := run(context.Background(), []string{"--installation", fixture.installationPath, "workspace", "inspect", "--workspace", "PSd"}, &stdout, &stderr)
+	if code != 2 || !strings.Contains(stderr.String(), "workspace") {
+		t.Fatalf("exit=%d stderr=%q", code, stderr.String())
+	}
+	assertDockerNotInvoked(t, fixture)
+}
+
 func TestRunSessionsMigrateRequiresExplicitConfirmationBeforeDocker(t *testing.T) {
 	fixture := newCLIFixture(t, "")
 	fixture.setProfile(t, "server")
@@ -855,3 +867,16 @@ func TestRenderWorkspaceHumanHidesRecoveryIdentity(t *testing.T) {
 		t.Fatalf("human output = %q", out.String())
 	}
 }
+
+func TestWorkspaceOperationalEnvelopeFailuresAreExitOne(t *testing.T) {
+	for _, err := range []error{compose.ErrOutputLimit, errors.New("invalid workspace result"), errors.New("invalid workspace request"), errors.New("invalid host export")} {
+		if workspaceUsageError(err) {
+			t.Errorf("classified operational error %q as usage", err)
+		}
+	}
+	for _, err := range []error{errors.New("unsafe output file"), errors.New("request exceeds limit"), errors.New("unsafe SQL input")} {
+		if !workspaceUsageError(err) {
+			t.Errorf("classified host error %q as operational", err)
+		}
+	}
+}

tools/thothctl/internal/compose/process_windows.go

@@ -3,9 +3,12 @@
 package compose
 
 import (
-	"golang.org/x/sys/windows"
+	"errors"
 	"os/exec"
 	"sync"
+	"unsafe"
+
+	"golang.org/x/sys/windows"
 )
 
 var ownedJobs = struct {
@@ -13,23 +16,76 @@ var ownedJobs = struct {
 	m map[*exec.Cmd]windows.Handle
 }{m: make(map[*exec.Cmd]windows.Handle)}
 
+// CREATE_SUSPENDED closes the registration race: no child code can create a
+// descendant until the real process HANDLE has been assigned to the job.
 func configureOwnedProcess(c *exec.Cmd) {
-	c.SysProcAttr = &windows.SysProcAttr{CreationFlags: windows.CREATE_NEW_PROCESS_GROUP}
+	c.SysProcAttr = &windows.SysProcAttr{CreationFlags: windows.CREATE_NEW_PROCESS_GROUP | windows.CREATE_SUSPENDED}
 }
+
 func registerOwnedProcess(c *exec.Cmd) error {
-	h, err := windows.CreateJobObject(nil, nil)
+	if c.Process == nil {
+		return errors.New("owned process has no process handle")
+	}
+	job, err := windows.CreateJobObject(nil, nil)
 	if err != nil {
 		return err
 	}
-	if err = windows.AssignProcessToJobObject(h, windows.Handle(c.Process.Pid)); err != nil {
-		windows.CloseHandle(h)
+	closeJob := true
+	defer func() {
+		if closeJob {
+			windows.CloseHandle(job)
+		}
+	}()
+	limits := windows.JOBOBJECT_EXTENDED_LIMIT_INFORMATION{}
+	limits.BasicLimitInformation.LimitFlags = windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE
+	if _, err = windows.SetInformationJobObject(job, windows.JobObjectExtendedLimitInformation, uintptr(unsafe.Pointer(&limits)), uint32(unsafe.Sizeof(limits))); err != nil {
 		return err
 	}
+	// c.Process.Pid is used only to obtain a genuine process HANDLE; a PID is
+	// never passed to AssignProcessToJobObject.
+	process, err := windows.OpenProcess(windows.PROCESS_SET_QUOTA|windows.PROCESS_TERMINATE|windows.PROCESS_QUERY_LIMITED_INFORMATION, false, uint32(c.Process.Pid))
+	if err != nil {
+		return err
+	}
+	defer windows.CloseHandle(process)
+	if err = windows.AssignProcessToJobObject(job, process); err != nil {
+		return err
+	}
+	thread, err := suspendedPrimaryThread(uint32(c.Process.Pid))
+	if err != nil {
+		_ = windows.TerminateJobObject(job, 1)
+		return err
+	}
+	_, resumeErr := windows.ResumeThread(thread)
+	windows.CloseHandle(thread)
+	if resumeErr != nil {
+		_ = windows.TerminateJobObject(job, 1)
+		return resumeErr
+	}
 	ownedJobs.Lock()
-	ownedJobs.m[c] = h
+	ownedJobs.m[c] = job
 	ownedJobs.Unlock()
+	closeJob = false
 	return nil
 }
+
+func suspendedPrimaryThread(pid uint32) (windows.Handle, error) {
+	snapshot, err := windows.CreateToolhelp32Snapshot(windows.TH32CS_SNAPTHREAD, 0)
+	if err != nil {
+		return 0, err
+	}
+	defer windows.CloseHandle(snapshot)
+	entry := windows.ThreadEntry32{Size: uint32(unsafe.Sizeof(windows.ThreadEntry32{}))}
+	err = windows.Thread32First(snapshot, &entry)
+	for err == nil {
+		if entry.OwnerProcessID == pid {
+			return windows.OpenThread(windows.THREAD_SUSPEND_RESUME, false, entry.ThreadID)
+		}
+		err = windows.Thread32Next(snapshot, &entry)
+	}
+	return 0, err
+}
+
 func releaseOwnedProcess(c *exec.Cmd) {
 	ownedJobs.Lock()
 	h := ownedJobs.m[c]

tools/thothctl/internal/compose/process_windows_test.go

@@ -0,0 +1,23 @@
+//go:build windows
+
+package compose
+
+import (
+	"golang.org/x/sys/windows"
+	"os/exec"
+	"testing"
+)
+
+func TestOwnedProcessConfigurationStartsSuspended(t *testing.T) {
+	cmd := exec.Command("cmd.exe")
+	configureOwnedProcess(cmd)
+	if cmd.SysProcAttr == nil || cmd.SysProcAttr.CreationFlags&windows.CREATE_SUSPENDED == 0 {
+		t.Fatal("owned Windows process must start suspended until job assignment")
+	}
+}
+
+func TestOwnedJobPolicyKillsDescendantsOnClose(t *testing.T) {
+	if windows.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE == 0 {
+		t.Fatal("owned job must use kill-on-close policy")
+	}
+}

tools/thothctl/internal/safeio/files_unix.go

@@ -90,7 +90,7 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err
 	if err != nil {
 		return ErrUnsafeFile
 	}
-	defer unix.Close(dir)
+	defer func() { unix.Close(dir) }()
 	for _, component := range components[:len(components)-1] {
 		next, err := unix.Openat(dir, component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0)
 		if err != nil {
@@ -147,7 +147,7 @@ func validateCanonicalOutputPath(path string) error {
 	if err != nil {
 		return ErrUnsafeFile
 	}
-	defer unix.Close(dir)
+	defer func() { unix.Close(dir) }()
 	for _, component := range components[:len(components)-1] {
 		next, err := unix.Openat(dir, component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0)
 		if err != nil {
@@ -173,7 +173,7 @@ func recheckUnixParents(components []string, retained []int) bool {
 	if err != nil {
 		return false
 	}
-	defer unix.Close(dir)
+	defer func() { unix.Close(dir) }()
 	for i, component := range components[:len(components)-1] {
 		next, e := unix.Openat(dir, component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0)
 		if e != nil {
@@ -195,7 +195,7 @@ func recheckUnixParentPath(components []string, retained int) bool {
 	if err != nil {
 		return false
 	}
-	defer unix.Close(dir)
+	defer func() { unix.Close(dir) }()
 	for _, component := range components {
 		next, e := unix.Openat(dir, component, unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0)
 		if e != nil {

tools/thothctl/internal/safeio/files_unix_test.go

@@ -30,3 +30,51 @@ func TestReadCanonicalRegularRejectsNamedPipeWithoutBlocking(t *testing.T) {
 		t.Fatalf("named pipe error = %v, want ErrUnsafeFile", err)
 	}
 }
+
+func TestCanonicalDescriptorOwnershipDoesNotLeakAcrossNestedOperations(t *testing.T) {
+	root, err := filepath.EvalSymlinks(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	nested := filepath.Join(root, "one", "two")
+	if err := os.MkdirAll(nested, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	input := filepath.Join(nested, "input.sql")
+	if err := os.WriteFile(input, []byte("select 1"), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	fdCount := func() int {
+		f, err := os.Open("/dev/fd")
+		if err != nil {
+			t.Fatal(err)
+		}
+		defer f.Close()
+		names, err := f.Readdirnames(-1)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return len(names)
+	}
+	baseline := fdCount()
+	for i := 0; i < 20; i++ {
+		if _, err := ReadCanonicalRegular(input, 1024); err != nil {
+			t.Fatal(err)
+		}
+		if err := validateCanonicalOutputPath(filepath.Join(nested, "out-"+string(rune('a'+i))+".yaml")); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if got := fdCount(); got > baseline+2 {
+		t.Fatalf("descriptor leak after read/validate: baseline=%d got=%d", baseline, got)
+	}
+	for i := 0; i < 20; i++ {
+		path := filepath.Join(nested, "write-"+string(rune('a'+i))+".yaml")
+		if err := writeCanonicalExclusive(path, []byte("ok"), 0o600); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if got := fdCount(); got > baseline+2 {
+		t.Fatalf("descriptor leak after writes: baseline=%d got=%d", baseline, got)
+	}
+}

tools/thothctl/internal/safeio/files_windows.go

@@ -122,7 +122,14 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err
 	if err := ValidateCanonicalPath(path); err != nil || len(contents) > 16<<20 || mode.Perm() == 0 || mode.Perm()&0o077 != 0 {
 		return ErrUnsafeFile
 	}
-	h, err := windows.CreateFile(windows.StringToUTF16Ptr(path), windows.GENERIC_WRITE, 0, nil, windows.CREATE_NEW, windows.FILE_ATTRIBUTE_NORMAL|windows.FILE_FLAG_OPEN_REPARSE_POINT, 0)
+	parent, retainedParents, err := openWindowsParents(path)
+	if err != nil {
+		return ErrUnsafeFile
+	}
+	defer closeWindowsHandles(retainedParents)
+	// Parent handles stay open with delete sharing denied until publication and
+	// identity recheck complete; this is the Windows equivalent of retained dirfds.
+	h, err := windows.CreateFile(windows.StringToUTF16Ptr(filepath.Join(parent, filepath.Base(path))), windows.GENERIC_WRITE, 0, nil, windows.CREATE_NEW, windows.FILE_ATTRIBUTE_NORMAL|windows.FILE_FLAG_OPEN_REPARSE_POINT, 0)
 	if err != nil {
 		return ErrUnsafeFile
 	}
@@ -132,6 +139,9 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err
 		return ErrUnsafeFile
 	}
 	defer f.Close()
+	if err := f.Chmod(mode); err != nil {
+		return ErrUnsafeFile
+	}
 	if _, err := f.Write(contents); err != nil {
 		return ErrUnsafeFile
 	}
@@ -153,26 +163,37 @@ func writeCanonicalExclusive(path string, contents []byte, mode fs.FileMode) err
 	return nil
 }
 
-func validateCanonicalOutputPath(path string) error {
-	if err := ValidateCanonicalPath(path); err != nil {
-		return err
-	}
+func openWindowsParents(path string) (string, []windows.Handle, error) {
 	volume := filepath.VolumeName(path)
 	root := volume + string(filepath.Separator)
 	components := strings.Split(strings.TrimPrefix(path, root), string(filepath.Separator))
 	if volume == "" || len(components) < 2 || components[0] == "" {
-		return ErrUnsafeFile
+		return "", nil, ErrUnsafeFile
 	}
 	current := root
+	parents := make([]windows.Handle, 0, len(components)-1)
 	for _, component := range components[:len(components)-1] {
 		current = filepath.Join(current, component)
 		h, err := openWindowsComponent(current, true)
 		if err != nil {
-			return ErrUnsafeFile
+			closeWindowsHandles(parents)
+			return "", nil, ErrUnsafeFile
 		}
-		windows.CloseHandle(h)
+		parents = append(parents, h)
 	}
-	if _, err := os.Lstat(filepath.Join(current, components[len(components)-1])); err == nil || !os.IsNotExist(err) {
+	return current, parents, nil
+}
+
+func validateCanonicalOutputPath(path string) error {
+	if err := ValidateCanonicalPath(path); err != nil {
+		return err
+	}
+	current, parents, err := openWindowsParents(path)
+	if err != nil {
+		return ErrUnsafeFile
+	}
+	defer closeWindowsHandles(parents)
+	if _, err := os.Lstat(filepath.Join(current, filepath.Base(path))); err == nil || !os.IsNotExist(err) {
 		return ErrUnsafeFile
 	}
 	return nil

tools/thothctl/internal/safeio/files_windows_test.go

@@ -66,3 +66,9 @@ func TestOpenWindowsComponentBlocksMutationWhileHandleIsRetained(t *testing.T) {
 		}
 	})
 }
+
+func TestWriteCanonicalExclusiveRequiresRestrictiveMode(t *testing.T) {
+	if err := writeCanonicalExclusive(`C:\\tmp\\thothctl-output.yaml`, []byte("x"), 0o640); err == nil {
+		t.Fatal("accepted non-restrictive output mode")
+	}
+}

tools/thothctl/internal/workspaceops/operations.go

@@ -404,7 +404,7 @@ func validateIngress(payload []byte, expected inputEnvelope) error {
 	var raw map[string]json.RawMessage
 	d := json.NewDecoder(bytes.NewReader(payload))
 	d.UseNumber()
-	if d.Decode(&raw) != nil {
+	if d.Decode(&raw) != nil || hasDuplicateJSONFields(payload) {
 		return errors.New("invalid workspace request")
 	}
 	var extra any
@@ -417,6 +417,12 @@ func validateIngress(payload []byte, expected inputEnvelope) error {
 	if dec.Decode(&supplied) != nil || supplied.SchemaVersion != 1 || supplied.Operation != expected.Operation || supplied.WorkspaceID != expected.WorkspaceID {
 		return errors.New("invalid workspace request")
 	}
+	// Presence is part of the command-derived envelope contract. Decoding into Go values
+	// alone would make an omitted field indistinguishable from an explicit zero/null value.
+	var generatedRaw map[string]json.RawMessage
+	if json.Unmarshal(mustJSON(expected), &generatedRaw) != nil || !sameJSONFieldSet(raw, generatedRaw) {
+		return errors.New("workspace request does not match command")
+	}
 	if !reflect.DeepEqual(supplied, expected) {
 		return errors.New("workspace request does not match command")
 	}
@@ -476,7 +482,7 @@ func Run(ctx context.Context, installation config.Installation, runner compose.R
 	}
 	var raw map[string]json.RawMessage
 	d := json.NewDecoder(strings.NewReader(cr.Stdout))
-	if d.Decode(&raw) != nil {
+	if d.Decode(&raw) != nil || hasDuplicateJSONFields([]byte(cr.Stdout)) {
 		return Result{}, errors.New("invalid workspace result")
 	}
 	var extra any
@@ -490,9 +496,20 @@ func Run(ctx context.Context, installation config.Installation, runner compose.R
 		if operationName(command) != "suggest-fks" {
 			return Result{}, errors.New("invalid workspace result")
 		}
-		if json.Unmarshal(h, &export) != nil {
+		var candidate hostExport
+		if hasDuplicateJSONFields(h) {
 			return Result{}, errors.New("invalid host export")
 		}
+		exportDecoder := json.NewDecoder(bytes.NewReader(h))
+		exportDecoder.DisallowUnknownFields()
+		if exportDecoder.Decode(&candidate) != nil {
+			return Result{}, errors.New("invalid host export")
+		}
+		var trailing any
+		if exportDecoder.Decode(&trailing) != io.EOF {
+			return Result{}, errors.New("invalid host export")
+		}
+		export = &candidate
 		delete(raw, "hostExport")
 	}
 	b, _ := json.Marshal(raw)
@@ -505,10 +522,13 @@ func Run(ctx context.Context, installation config.Installation, runner compose.R
 	if e = validateResult(result, env.WorkspaceID, operationName(command)); e != nil {
 		return Result{}, e
 	}
+	if !resultExitMatches(result.Status, cr.ExitCode) {
+		return Result{}, runErrOr(runErr, "workspace result status does not match child exit")
+	}
 	if runErr != nil {
-		if result.Status == "blocked" && cr.ExitCode == 3 {
-		} else if result.Status == "failed" && cr.ExitCode == 1 {
-		} else {
+		// Exit 1/3 are represented by the validated public result; callers still
+		// render it and select the corresponding process exit code.
+		if result.Status != "blocked" && result.Status != "failed" {
 			return Result{}, runErr
 		}
 	}
@@ -552,9 +572,6 @@ func candidateBoundToResult(x hostExport, result Result) bool {
 }
 
 func publishCandidate(x *hostExport, result Result, path string) error {
-	if path == "" {
-		return nil
-	}
 	if x.MediaType != "application/yaml" && x.MediaType != "text/yaml" {
 		return errors.New("invalid candidate export")
 	}
@@ -575,6 +592,9 @@ func publishCandidate(x *hostExport, result Result, path string) error {
 	if result.RunID == "" || !runIDPattern.MatchString(result.RunID) {
 		return errors.New("invalid candidate identity")
 	}
+	if path == "" {
+		return nil
+	}
 	if e = safeio.WriteCanonicalExclusive(path, b, 0o600); e != nil {
 		return errors.New("unsafe output file")
 	}
@@ -606,8 +626,89 @@ func validateResult(r Result, workspace, operation string) error {
 	if r.Status != "blocked" && r.Code == CodeRegistryBootstrapRecoveryConflict {
 		return errors.New("invalid workspace result")
 	}
+	if (r.Status == "succeeded" || r.Status == "unchanged" || r.Status == "dry_run") && r.Code != "ok" {
+		return errors.New("invalid workspace result")
+	}
+	if r.Status == "failed" && (r.Code == "ok" || r.Code == "manual_review_required" || r.Code == "evidence_materialization_required" || r.Code == "preprocessing_conflict" || r.Code == "preprocessing_resume_mismatch") {
+		return errors.New("invalid workspace result")
+	}
 	return nil
 }
+
+func resultExitMatches(status string, exitCode int) bool {
+	switch status {
+	case "succeeded", "unchanged", "dry_run":
+		return exitCode == 0
+	case "blocked":
+		return exitCode == 3
+	case "failed":
+		return exitCode == 1
+	default:
+		return false
+	}
+}
+
+func hasDuplicateJSONFields(payload []byte) bool {
+	decoder := json.NewDecoder(bytes.NewReader(payload))
+	decoder.UseNumber()
+	if !scanJSONValue(decoder) {
+		return true
+	}
+	_, err := decoder.Token()
+	return err != io.EOF
+}
+
+func scanJSONValue(decoder *json.Decoder) bool {
+	token, err := decoder.Token()
+	if err != nil {
+		return false
+	}
+	delimiter, isDelimiter := token.(json.Delim)
+	if !isDelimiter {
+		return true
+	}
+	switch delimiter {
+	case '{':
+		seen := map[string]bool{}
+		for decoder.More() {
+			keyToken, keyErr := decoder.Token()
+			key, ok := keyToken.(string)
+			if keyErr != nil || !ok || seen[key] || !scanJSONValue(decoder) {
+				return false
+			}
+			seen[key] = true
+		}
+		_, err = decoder.Token()
+		return err == nil
+	case '[':
+		for decoder.More() {
+			if !scanJSONValue(decoder) {
+				return false
+			}
+		}
+		_, err = decoder.Token()
+		return err == nil
+	default:
+		return false
+	}
+}
+
+func sameJSONFieldSet(a, b map[string]json.RawMessage) bool {
+	if len(a) != len(b) {
+		return false
+	}
+	for key := range a {
+		if _, ok := b[key]; !ok {
+			return false
+		}
+	}
+	return true
+}
+
+func mustJSON(v any) []byte {
+	b, _ := json.Marshal(v)
+	return b
+}
 func DigestBytes(b []byte) string { s := sha256.Sum256(b); return "sha256:" + hex.EncodeToString(s[:]) }
 func validStatus(v string) bool {
 	switch v {

tools/thothctl/internal/workspaceops/operations_test.go

@@ -1,7 +1,9 @@
 package workspaceops
 
 import (
+	"bytes"
 	"encoding/base64"
+	"fmt"
 	"os"
 	"path/filepath"
 	"reflect"
@@ -132,3 +134,164 @@ func TestRunPublishesOnlyVerifiedCandidateExport(t *testing.T) {
 		t.Fatalf("candidate = %q, %v", b, err)
 	}
 }
+
+func validWorkspaceResult(status, code string) Result {
+	return Result{SchemaVersion: 1, Status: status, Code: code, WorkspaceID: "psd", WorkspaceRevision: strings.Repeat("0", 40), DescriptorBlob: strings.Repeat("a", 40), Operation: "evidence", CompletedStages: []string{}}
+}
+
+func TestValidateIngressRejectsExplicitOmittedEvidenceZeroFields(t *testing.T) {
+	command := EvidenceRequest{WorkspaceID: "psd"}
+	expected, _, err := makeInput(command)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for _, payload := range []string{
+		`{"schemaVersion":1,"operation":"evidence","workspaceId":"psd","resume":""}`,
+		`{"schemaVersion":1,"operation":"evidence","workspaceId":"psd","workspaceId":"psd"}`,
+		`{"schemaVersion":1,"operation":"evidence","workspaceId":"psd","dryRun":false}`,
+		`{"schemaVersion":1,"operation":"evidence","workspaceId":"psd","resume":null}`,
+	} {
+		if err := validateIngress([]byte(payload), expected); err == nil {
+			t.Errorf("accepted non-derived envelope %s", payload)
+		}
+	}
+}
+
+func TestValidateIngressRequiresExactDerivedFieldSetForOptionalInputs(t *testing.T) {
+	command := SuggestFksRequest{WorkspaceID: "psd"}
+	expected, _, err := makeInput(command)
+	if err != nil {
+		t.Fatal(err)
+	}
+	payload := `{"schemaVersion":1,"operation":"suggest-fks","workspaceId":"psd","sql":null,"assume":[]}`
+	if err := validateIngress([]byte(payload), expected); err == nil {
+		t.Fatal("accepted explicit null/empty optional fields omitted by command")
+	}
+}
+
+func TestPublishCandidateValidatesInternalExportWithoutOutput(t *testing.T) {
+	candidate := []byte("candidates: []\n")
+	digest := DigestBytes(candidate)
+	result := validWorkspaceResult("succeeded", "ok")
+	result.Operation = "suggest-fks"
+	result.RunID = strings.Repeat("b", 32)
+	result.ArtifactIdentities = []ArtifactIdentity{{Kind: "fk-candidates", Digest: digest}}
+	for _, export := range []hostExport{
+		{MediaType: "application/json", SHA256: digest, ContentBase64: base64.StdEncoding.EncodeToString(candidate)},
+		{MediaType: "application/yaml", SHA256: digest, ContentBase64: base64.StdEncoding.EncodeToString(candidate)},
+	} {
+		if export.MediaType == "application/json" {
+			if err := publishCandidate(&export, result, ""); err == nil {
+				t.Fatal("accepted invalid media type without output")
+			}
+		} else if err := publishCandidate(&export, result, ""); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func TestValidateResultBindsStatusCodeAndExit(t *testing.T) {
+	cases := []struct {
+		status, code string
+		exit         int
+		valid        bool
+	}{
+		{"succeeded", "ok", 0, true}, {"unchanged", "ok", 0, true}, {"dry_run", "ok", 0, true},
+		{"blocked", "manual_review_required", 3, true}, {"failed", "workspace_not_found", 1, true},
+		{"blocked", "manual_review_required", 0, false}, {"failed", "workspace_not_found", 0, false},
+		{"succeeded", "workspace_not_found", 0, false}, {"failed", "ok", 1, false},
+	}
+	for _, tc := range cases {
+		r := validWorkspaceResult(tc.status, tc.code)
+		err := validateResult(r, "psd", "evidence")
+		got := err == nil && resultExitMatches(tc.status, tc.exit)
+		if got != tc.valid {
+			t.Errorf("%s/%s exit %d valid=%v err=%v", tc.status, tc.code, tc.exit, tc.valid, err)
+		}
+	}
+}
+
+func TestRunRejectsUnknownHostExportFieldWithoutOutput(t *testing.T) {
+	d := t.TempDir()
+	candidate := []byte("candidates: []\n")
+	digest := DigestBytes(candidate)
+	response := `{"schemaVersion":1,"status":"succeeded","code":"ok","workspaceId":"psd","workspaceRevision":"` + strings.Repeat("0", 40) + `","descriptorBlob":"` + strings.Repeat("a", 40) + `","operation":"suggest-fks","runId":"` + strings.Repeat("b", 32) + `","completedStages":[],"artifactIdentities":[{"kind":"fk-candidates","digest":"` + digest + `"}],"hostExport":{"mediaType":"application/yaml","sha256":"` + digest + `","contentBase64":"` + base64.StdEncoding.EncodeToString(candidate) + `","extra":1}}`
+	fake := filepath.Join(d, "docker")
+	if err := os.WriteFile(fake, []byte("#!/bin/sh\ncat >/dev/null\nprintf '%s' '"+response+"'\n"), 0o700); err != nil {
+		t.Fatal(err)
+	}
+	cmd := SuggestFksRequest{WorkspaceID: "psd"}
+	if _, err := Run(context.Background(), config.Installation{ProjectDirectory: d}, compose.NewRunner(fake), cmd, nil); err == nil {
+		t.Fatal("accepted unknown hostExport field")
+	}
+}
+
+func TestPublishCandidateRejectsOversizedExportWithoutOutput(t *testing.T) {
+	candidate := bytes.Repeat([]byte("y"), maxCandidate+1)
+	digest := DigestBytes(candidate)
+	x := hostExport{MediaType: "application/yaml", SHA256: digest, ContentBase64: base64.StdEncoding.EncodeToString(candidate)}
+	r := validWorkspaceResult("succeeded", "ok")
+	r.Operation, r.RunID = "suggest-fks", strings.Repeat("b", 32)
+	if err := publishCandidate(&x, r, ""); err == nil {
+		t.Fatal("accepted oversized export without output")
+	}
+}
+
+func TestRunAcceptsOnlyMatchingChildExitForStatus(t *testing.T) {
+	for _, tc := range []struct {
+		name, status, code string
+		exit               int
+		wantErr            bool
+	}{
+		{"blocked-exit3", "blocked", "manual_review_required", 3, false},
+		{"blocked-exit0", "blocked", "manual_review_required", 0, true},
+		{"failed-exit1", "failed", "workspace_not_found", 1, false},
+		{"failed-exit0", "failed", "workspace_not_found", 0, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			d := t.TempDir()
+			response := fmt.Sprintf(`{"schemaVersion":1,"status":"%s","code":"%s","workspaceId":"psd","workspaceRevision":"%s","descriptorBlob":"%s","operation":"evidence","completedStages":[]}`,
+				tc.status, tc.code, strings.Repeat("0", 40), strings.Repeat("a", 40))
+			fake := filepath.Join(d, "docker")
+			if err := os.WriteFile(fake, []byte("#!/bin/sh\ncat >/dev/null\nprintf '%s' '"+response+"'\nexit "+fmt.Sprint(tc.exit)+"\n"), 0o700); err != nil {
+				t.Fatal(err)
+			}
+			_, err := Run(context.Background(), config.Installation{ProjectDirectory: d}, compose.NewRunner(fake), EvidenceRequest{WorkspaceID: "psd"}, nil)
+			if (err != nil) != tc.wantErr {
+				t.Fatalf("error=%v wantErr=%v", err, tc.wantErr)
+			}
+		})
+	}
+}
+
+func TestParseWorkspaceRejectsDuplicateMissingAndUnknownOptions(t *testing.T) {
+	validResume := strings.Repeat("0", 32)
+	for _, argv := range [][]string{
+		{"workspace", "inspect", "--json", "--json", "--workspace", "psd"},
+		{"workspace", "inspect", "--workspace", "psd", "--workspace", "psd"},
+		{"workspace", "preprocess", "dwh", "--workspace", "psd", "--resume", validResume, "--resume", validResume},
+		{"workspace", "schema", "suggest-fks", "--workspace", "psd", "--output", "a", "--output", "b"},
+		{"workspace", "inspect", "--json", "--unknown", "x", "--workspace", "psd"},
+		{"workspace", "inspect", "--json"},
+	} {
+		if _, err := ParseWorkspaceCommand(argv); err == nil {
+			t.Errorf("accepted invalid argv %v", argv)
+		}
+	}
+}
+
+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 {
+		t.Fatal("accepted oversized assumption")
+	}
+	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))
+	}
+	if _, err := ParseWorkspaceCommand(args); err == nil {
+		t.Fatal("accepted 33 SQL files")
+	}
+}