fix(evidence): bound reports and fail failed jobs

harness/tests/test_corpus_pipeline.py

@@ -339,7 +339,27 @@ def test_pipeline_result_dump_does_not_deepcopy_frozen_metadata():
     payload = PipelineResult(
         "succeeded", None, False, (), (), (), manifest,
     ).model_dump(mode="json")
-    assert payload["manifest"]["metadata"] == {"nested": {"value": ["safe"]}}
+    assert payload["manifest_id"] is None
+    assert "manifest" not in payload
+
+
+def test_pipeline_result_public_dump_is_bounded_and_excludes_evidence_content(tmp_path):
+    import json
+
+    result = pipeline(
+        tmp_path, Source([(item("one", "a"), "SENSITIVE EVIDENCE CONTENT")])
+    ).run()
+    payload = result.model_dump(mode="json")
+    encoded = json.dumps(payload)
+    assert "SENSITIVE EVIDENCE CONTENT" not in encoded
+    assert "documents" not in payload and "chunks" not in payload
+    large = PipelineResult(
+        "failed", None, False,
+        tuple(f"fs:item-{index}" for index in range(1000)), (), (), result.manifest,
+    ).model_dump(mode="json")
+    assert len(large["changed"]) == 100
+    assert large["counts"]["changed"] == 1000
+    assert len(json.dumps(large)) < 25_000
 
 
 def test_reused_corpus_root_rejects_workspace_rename_before_any_mutation(tmp_path):

harness/tests/test_preprocess_cli.py

@@ -32,6 +32,46 @@ def test_preprocess_failure_is_structured_and_nonzero(monkeypatch, tmp_path):
     assert "secret detail" not in response.output
 
 
+def test_preprocess_failed_job_report_is_sanitized_json_and_nonzero(monkeypatch, tmp_path):
+    import tht.cli.preprocess_cmd as command
+
+    result = SimpleNamespace(model_dump=lambda mode=None: {
+        "status": "failed", "run_id": "a" * 32, "published": False,
+        "generation": "gen:" + "b" * 32, "changed": ["fs:one"],
+    })
+    monkeypatch.setattr(command, "run_from_config", lambda *args, **kwargs: result)
+    response = CliRunner().invoke(
+        app, ["preprocess", "evidence", "--json", "-c", str(tmp_path / "workspace.yaml")]
+    )
+    assert response.exit_code == 1
+    payload = json.loads(response.output)
+    assert payload["status"] == "failed"
+    assert payload["error"] == "preprocessing job failed"
+    assert "traceback" not in response.output.lower()
+
+
+def test_preprocess_real_failed_stage_result_exits_nonzero(monkeypatch, tmp_path):
+    import tht.cli.preprocess_cmd as command
+    from test_corpus_pipeline import Source, item, pipeline
+
+    result = pipeline(
+        tmp_path, Source([(item("one", "a"), RuntimeError("SENSITIVE EVIDENCE secret"))])
+    ).run_as_job(
+        workspace_id="demo", workspace_root=tmp_path,
+        config_fingerprint="sha256:" + "1" * 64,
+        input_fingerprint="sha256:" + "2" * 64,
+    )
+    assert result.status == "failed"
+    monkeypatch.setattr(command, "run_from_config", lambda *args, **kwargs: result)
+    response = CliRunner().invoke(
+        app, ["preprocess", "evidence", "--json", "-c", str(tmp_path / "workspace.yaml")]
+    )
+    assert response.exit_code == 1
+    assert json.loads(response.output)["status"] == "failed"
+    assert "SENSITIVE EVIDENCE" not in response.output
+    assert "secret" not in response.output
+
+
 def test_preprocess_resume_rejects_generation_id_before_configuration(monkeypatch, tmp_path):
     import tht.cli.preprocess_cmd as command
 

harness/tht/cli/preprocess_cmd.py

@@ -159,6 +159,13 @@ def evidence_cmd(
             typer.secho("ERRORE: preprocessing failed", fg=typer.colors.RED, err=True)
         raise typer.Exit(code=1) from None
     payload = result.model_dump(mode="json")
+    if payload.get("status") != "succeeded":
+        payload["error"] = "preprocessing job failed"
+        if json_output:
+            typer.echo(json.dumps(payload, ensure_ascii=False, sort_keys=True))
+        else:
+            typer.secho("ERRORE: preprocessing job failed", fg=typer.colors.RED, err=True)
+        raise typer.Exit(code=1)
     if json_output:
         typer.echo(json.dumps(payload, ensure_ascii=False, sort_keys=True))
     else:

harness/tht/corpus/pipeline.py

@@ -50,14 +50,24 @@ class PipelineResult:
     resumed_from: str | None = None
 
     def model_dump(self, mode=None):
+        def bounded(values: tuple[str, ...]) -> list[str]:
+            return [value[:200] for value in values[:100]]
+
         return {
             "status": self.status,
             "generation": self.generation,
             "published": self.published,
-            "changed": list(self.changed),
-            "unchanged": list(self.unchanged),
-            "removed": list(self.removed),
-            "manifest": self.manifest.model_dump(mode="json"),
+            "changed": bounded(self.changed),
+            "unchanged": bounded(self.unchanged),
+            "removed": bounded(self.removed),
+            "counts": {
+                "changed": len(self.changed),
+                "unchanged": len(self.unchanged),
+                "removed": len(self.removed),
+                "documents": len(self.manifest.documents),
+                "chunks": len(self.manifest.chunks),
+            },
+            "manifest_id": self.manifest.manifest_id,
             "run_id": self.run_id,
             "resumed_from": self.resumed_from,
         }