fix(noema): bind sidecar evidence to current producer

.github/workflows/noema-review.yml

@@ -910,6 +910,48 @@ jobs:
             --private-target "$PRIVATE_TARGET"
           echo 'selected=true' >>"$GITHUB_OUTPUT"
 
+      - name: Initialize current Noema sidecar evidence
+        id: noema_evidence
+        if: env.PR_NUMBER != '' && steps.live_draft.outputs.live_draft != 'true' && steps.personal_route.outputs.selected != 'true'
+        run: |
+          set -euo pipefail
+          # Separate setup from the producer: unsafe paths fail before its outcome
+          # can admit an upload. Clear only these two files, before early startup.
+          python3 -I <<'PY'
+          import os
+          import stat
+          from pathlib import Path
+
+          workspace_text = os.environ["GITHUB_WORKSPACE"]
+          workspace = Path(workspace_text)
+          if (
+              not workspace.is_absolute()
+              or ".." in workspace.parts
+              or workspace.resolve(strict=True) != workspace
+              or not workspace.is_dir()
+          ):
+              raise SystemExit("Noema evidence workspace must be an unaliased directory")
+          evidence = workspace / "strix_runs"
+          if evidence.is_symlink() or (evidence.exists() and not evidence.is_dir()):
+              raise SystemExit("Noema evidence parent must be a non-symlink directory")
+          owned = (
+              evidence / "contextual-orchestrator-preflight.json",
+              evidence / "contextual-orchestrator-sidecar.stderr.log",
+          )
+          # Validate both before deleting either; reject dangling links, special
+          # files and hardlink aliases as well as nonregular evidence paths.
+          for path in owned:
+              try:
+                  metadata = path.lstat()
+              except FileNotFoundError:
+                  continue
+              if not stat.S_ISREG(metadata.st_mode) or metadata.st_nlink != 1:
+                  raise SystemExit("Noema evidence must be regular, single-link files")
+          evidence.mkdir(mode=0o700, exist_ok=True)
+          for path in owned:
+              path.unlink(missing_ok=True)
+          PY
+
       - name: Provision contextual-orchestrator review sidecar
         id: noema_sidecar
         if: env.PR_NUMBER != '' && steps.live_draft.outputs.live_draft != 'true' && steps.personal_route.outputs.selected != 'true'
@@ -1016,7 +1058,10 @@ jobs:
           fi
 
       - name: Upload contextual-orchestrator sidecar evidence
-        if: always() && env.PR_NUMBER != ''
+        # A live Draft never provisions the sidecar. Retain only current producer
+        # terminal diagnostics; absent/skipped outcomes cannot publish stale files.
+        # Non-Draft upload failures still fail the required review job.
+        if: always() && env.PR_NUMBER != '' && steps.live_draft.outputs.live_draft != 'true' && (steps.noema_sidecar.outcome == 'success' || steps.noema_sidecar.outcome == 'failure' || steps.noema_sidecar.outcome == 'cancelled')
         uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
         with:
           name: noema-sidecar-evidence

tests/test_noema_draft_admission_before_sidecar.py

@@ -19,10 +19,19 @@
 import textwrap
 from pathlib import Path
 
+import pytest
+import yaml
+
 from tests.test_required_workflow_queue_contract import workflow_step, workflow_text
 
 DRAFT_STEP = "Check live pull request draft state before sidecar provisioning"
 DRAFT_GATE = "steps.live_draft.outputs.live_draft != 'true'"
+EXPECTED_UPLOAD_CONDITION = (
+    "always() && env.PR_NUMBER != '' && " + DRAFT_GATE
+    + " && (steps.noema_sidecar.outcome == 'success'"
+    + " || steps.noema_sidecar.outcome == 'failure'"
+    + " || steps.noema_sidecar.outcome == 'cancelled')"
+)
 REVIEWER_TOKEN = (
     "GH_TOKEN: ${{ env.TARGET_REPOSITORY == 'ContextualWisdomLab/k-csap-skills' && "
     "steps.noema_github_app_token.outputs.token || (env.TARGET_REPOSITORY != "
@@ -49,6 +58,114 @@ def _step_index(job: str, name: str) -> int:
     return job.index(f"      - name: {name}\n")
 
 
+def _parsed_noema_job(workflow: str) -> dict:
+    """Read the actual YAML job, retaining GitHub's literal ``on`` key."""
+    return yaml.load(workflow, Loader=yaml.BaseLoader)["jobs"]["noema-review"]
+
+
+def _sidecar_upload_is_fail_closed(workflow: str) -> bool:
+    """Bind the exact gate to one case-normalized action, never a display-name decoy."""
+    uploads = [
+        step for step in _parsed_noema_job(workflow)["steps"]
+        if str(step.get("uses", "")).casefold().startswith("actions/upload-artifact@")
+    ]
+    if len(uploads) != 1:
+        return False
+    upload = uploads[0]
+    return (
+        upload.get("name") == "Upload contextual-orchestrator sidecar evidence"
+        and upload.get("if") == EXPECTED_UPLOAD_CONDITION
+        and upload.get("with", {}).get("name") == "noema-sidecar-evidence"
+        and "continue-on-error" not in upload
+    )
+
+
+def _noema_review_job_is_fail_closed(workflow: str) -> bool:
+    """The review job itself must propagate upload and producer failures."""
+    return "continue-on-error" not in _parsed_noema_job(workflow)
+
+
+def _upload_admitted(workflow: str, pr: str, draft: str, outcome: str) -> bool:
+    """Evaluate the actual condition's bounded grammar; unknown expressions fail."""
+    uploads = [
+        step for step in _parsed_noema_job(workflow)["steps"]
+        if str(step.get("uses", "")).casefold().startswith("actions/upload-artifact@")
+    ]
+    assert len(uploads) == 1
+    terms = {
+        "always()": True,
+        "env.PR_NUMBER != ''": bool(pr),
+        DRAFT_GATE: draft != "true",
+        "steps.noema_sidecar.outcome == 'success'": outcome == "success",
+        "steps.noema_sidecar.outcome == 'failure'": outcome == "failure",
+        "steps.noema_sidecar.outcome == 'cancelled'": outcome == "cancelled",
+    }
+    values = []
+    for part in uploads[0]["if"].split(" && "):
+        if part.startswith("(") and part.endswith(")"):
+            alternatives = part[1:-1].split(" || ")
+            assert all(term in terms for term in alternatives), part
+            values.append(any(terms[term] for term in alternatives))
+        else:
+            assert part in terms, part
+            values.append(terms[part])
+    return all(values)
+
+
+@pytest.mark.parametrize("pr", ["", "2565"])
+@pytest.mark.parametrize("draft", ["true", "false", ""])
+@pytest.mark.parametrize("outcome", ["", "skipped", "unknown", "success", "failure", "cancelled"])
+def test_upload_admission_requires_current_terminal_producer(pr: str, draft: str, outcome: str) -> None:
+    """Absent/skipped producers cannot publish stale files; terminal diagnostics survive."""
+    admitted = _upload_admitted(workflow_text("noema-review.yml"), pr, draft, outcome)
+    assert admitted == (
+        bool(pr) and draft != "true" and outcome in {"success", "failure", "cancelled"}
+    ), (pr, draft, outcome)
+
+
+def test_upload_action_identity_oracle_rejects_decoys_and_failure_bypasses() -> None:
+    """Keep the owner's unique casefold oracle and quoted/spaced failure controls."""
+    workflow = workflow_text("noema-review.yml")
+    assert _sidecar_upload_is_fail_closed(workflow)
+    assert _noema_review_job_is_fail_closed(workflow)
+    assert not _sidecar_upload_is_fail_closed(
+        workflow.replace(EXPECTED_UPLOAD_CONDITION, EXPECTED_UPLOAD_CONDITION + " || true", 1)
+    )
+    marker = "      - name: Upload contextual-orchestrator sidecar evidence\n"
+    for bypass in ("continue-on-error: true", '"continue-on-error": true', "continue-on-error : true"):
+        assert not _sidecar_upload_is_fail_closed(
+            workflow.replace(marker, marker + "        " + bypass + "\n", 1)
+        ), bypass
+        assert not _noema_review_job_is_fail_closed(
+            workflow.replace("  noema-review:\n", "  noema-review:\n    " + bypass + "\n", 1)
+        ), bypass
+    run_decoy = marker + f"        if: {EXPECTED_UPLOAD_CONDITION}\n        run: 'true'\n\n"
+    assert _sidecar_upload_is_fail_closed(workflow.replace(marker, run_decoy + marker, 1))
+    assert not _sidecar_upload_is_fail_closed(
+        workflow.replace(marker, run_decoy + marker + "        continue-on-error: true\n", 1)
+    )
+    action = next(
+        step["uses"] for step in _parsed_noema_job(workflow)["steps"]
+        if str(step.get("uses", "")).casefold().startswith("actions/upload-artifact@")
+    )
+    mixed_case = workflow.replace(
+        f"uses: {action}", f"uses: Actions/Upload-Artifact@{action.split('@', 1)[1]}", 1
+    )
+    assert mixed_case != workflow
+    assert _sidecar_upload_is_fail_closed(mixed_case)
+    action_decoy = (
+        marker + f"        if: {EXPECTED_UPLOAD_CONDITION}\n        uses: {action}\n"
+        + "        with:\n          name: noema-sidecar-evidence\n          path: CHANGELOG.md\n\n"
+    )
+    assert not _sidecar_upload_is_fail_closed(mixed_case.replace(marker, action_decoy + marker, 1))
+    assert not _sidecar_upload_is_fail_closed(mixed_case.replace(
+        marker, action_decoy + marker + "        continue-on-error: true\n", 1
+    ))
+    assert _noema_review_job_is_fail_closed(workflow.replace(
+        "  continue-noema-transport:\n", "  continue-noema-transport:\n    continue-on-error: true\n", 1
+    ))
+
+
 def test_live_draft_check_runs_after_head_validation_and_before_sidecar() -> None:
     """The draft decision sits between live-head validation and model provisioning."""
     job = _noema_job()
@@ -121,9 +238,8 @@ def test_downstream_publication_treats_unset_prepare_outputs_as_skipped() -> Non
     assert "needs.noema-review.result == 'failure'" in continuation
     assert "needs.noema-review.outputs.transport_capacity_unavailable == 'true'" in continuation
     assert "needs.noema-review.outputs.transport_retry_eligible == 'true'" in continuation
-    assert "if: always() && env.PR_NUMBER != ''" in workflow_step(
-        workflow, "Upload contextual-orchestrator sidecar evidence"
-    )
+    assert _noema_review_job_is_fail_closed(workflow)
+    assert _sidecar_upload_is_fail_closed(workflow)
 
 
 def test_trigger_types_are_unchanged_by_the_runtime_draft_check() -> None:
@@ -141,6 +257,7 @@ def _run_draft_step(tmp_path: Path, gh_body: str, gh_token: str = "synthetic-tok
     script = textwrap.dedent(
         workflow_step(workflow_text("noema-review.yml"), DRAFT_STEP).split("        run: |\n", 1)[1]
     )
+    script = script.replace("/tmp/noema-live-draft-error", str(tmp_path / "live-draft-error"))
     fake_gh = tmp_path / "gh"
     fake_gh.write_text(f"#!/usr/bin/env bash\n{gh_body}\n", encoding="utf-8")
     fake_gh.chmod(0o755)