feat(P25): deploy outputs (SSM + PR comment) + error reporting via Lambda + stage comments

---ci---
project: acdl
phase: 25
milestone: v1.7
status: execute
---/ci---

.gitea/workflows/ci.yml

@@ -37,6 +37,7 @@ jobs:
           python3 -m py_compile \
             core/confidence_signal.py \
             core/outbox_writer.py \
+            core/output_publisher.py \
             core/contract_resolver.py \
             core/lambda/contract_ingestor.py \
             adapters/terraform/adapter.py \

.gitea/workflows/deploy.yml

@@ -112,6 +112,26 @@ jobs:
           esac
           bash platform/scripts/run_platform.sh $MODE_FLAG "${{ inputs.contract }}"
 
+      - name: Post stage summary comment to PR
+        if: success() && github.event_name == 'pull_request'
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GITHUB_REPOSITORY: ${{ github.repository }}
+          GITHUB_REF: ${{ github.ref }}
+        run: |
+          bash platform/scripts/post_stage_comment.sh deploy pass '{"mode":"${{ inputs.mode }}","runId":"${{ github.run_id }}"}'
+
+      - name: Report error to platform team (on failure)
+        if: failure()
+        env:
+          AWS_DEFAULT_REGION: us-east-1
+        run: |
+          aws lambda invoke-function-url \
+            --function-url "${{ secrets.ACDL_LAMBDA_URL }}" \
+            --cli-binary-format raw-in-base64-out \
+            --payload "$(python3 -c "import json,os; print(json.dumps({'action':'report_error','consumerRepo':os.environ.get('GITHUB_REPOSITORY',''),'contractId':'${{ github.run_id }}','error':'Deploy pipeline failed. See run logs.','runUrl':'${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}','environment':'dev'}))")" \
+            /dev/null || true
+
       - name: Upload emitted Terraform
         uses: actions/upload-artifact@v4
         with:

.github/workflows/ci.yml

@@ -37,6 +37,7 @@ jobs:
           python3 -m py_compile \
             core/confidence_signal.py \
             core/outbox_writer.py \
+            core/output_publisher.py \
             core/contract_resolver.py \
             core/lambda/contract_ingestor.py \
             adapters/terraform/adapter.py \

.github/workflows/deploy.yml

@@ -112,6 +112,26 @@ jobs:
           esac
           bash platform/scripts/run_platform.sh $MODE_FLAG "${{ inputs.contract }}"
 
+      - name: Post stage summary comment to PR
+        if: success() && github.event_name == 'pull_request'
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+          GITHUB_REPOSITORY: ${{ github.repository }}
+          GITHUB_REF: ${{ github.ref }}
+        run: |
+          bash platform/scripts/post_stage_comment.sh deploy pass '{"mode":"${{ inputs.mode }}","runId":"${{ github.run_id }}"}'
+
+      - name: Report error to platform team (on failure)
+        if: failure()
+        env:
+          AWS_DEFAULT_REGION: us-east-1
+        run: |
+          aws lambda invoke-function-url \
+            --function-url "${{ secrets.ACDL_LAMBDA_URL }}" \
+            --cli-binary-format raw-in-base64-out \
+            --payload "$(python3 -c "import json,os; print(json.dumps({'action':'report_error','consumerRepo':os.environ.get('GITHUB_REPOSITORY',''),'contractId':'${{ github.run_id }}','error':'Deploy pipeline failed. See run logs.','runUrl':'${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}','environment':'dev'}))")" \
+            /dev/null || true
+
       - name: Upload emitted Terraform
         uses: actions/upload-artifact@v4
         with:

core/lambda/contract_ingestor.py

@@ -5,8 +5,9 @@ communication, D-051). Accepts { consumerRepo, contractId, contract,
 environment, action } and writes contracts to DynamoDB table acdl-contracts
 (PK consumerRepo, SK contractId#submittedAt).
 
-The report_error action (D-055) is prepared as a stub in this phase; the
-GitHub issue creation is implemented in Phase 25.
+The report_error action (D-055) creates a GitHub issue on the platform repo
+via the GitHub API, using a token from Secrets Manager. It is idempotent: if
+an open issue with the same title exists, it comments rather than duplicating.
 
 Cross-account: the Lambda's Function URL uses IAM auth; the consumer's
 deploy role (granted during onboarding) invokes it via SigV4-signed
@@ -71,17 +72,109 @@ def _submit_contract(payload):
 
 
 def _report_error(payload):
-    # Phase 25 implements the GitHub issue creation.
-    # This stub validates the payload and returns a prepared status.
+    """Create a GitHub issue on the platform repo for a deploy failure (D-055).
+
+    Uses the GitHub token from Secrets Manager. Idempotent: if an open
+    issue with the same title exists, comments on it rather than duplicating.
+    """
+    import urllib.request
+
     required = ["consumerRepo", "contractId", "error"]
     for field in required:
         if field not in payload:
             raise ValueError(f"report_error requires '{field}'")
-    return {
-        "status": "error_report_prepared",
-        "contractId": payload["contractId"],
-        "action": "report_error",
-    }
+
+    consumer_repo = payload["consumerRepo"]
+    contract_id = payload["contractId"]
+    error = payload.get("error", "unknown error")
+    run_url = payload.get("runUrl", "")
+    stack_trace = payload.get("stackTrace", "")[:2000]  # truncate
+
+    # Get the GitHub token from Secrets Manager
+    secrets = _get_secrets_client()
+    try:
+        secret_response = secrets.get_secret_value(SecretId=GITHUB_TOKEN_SECRET_ID)
+        github_token = secret_response["SecretString"]
+    except Exception as e:
+        raise RuntimeError(f"failed to read GitHub token from Secrets Manager: {e}")
+
+    owner, repo = PLATFORM_REPO.split("/")
+    title = f"[ACDL-ALERT] Deploy failure: {consumer_repo} / {contract_id}"
+
+    # Check for an existing open issue with the same title (idempotency)
+    search_url = (
+        f"https://api.github.com/search/issues?q=repo:{owner}/{repo}"
+        f"+is:issue+is:open+in:title+%22{contract_id}%22"
+    )
+    req = urllib.request.Request(search_url)
+    req.add_header("Authorization", f"token {github_token}")
+    req.add_header("Accept", "application/vnd.github+json")
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            search_result = json.loads(resp.read())
+        existing = search_result.get("items", [])
+    except Exception:
+        existing = []
+
+    body = f"""## Deploy Failure Report
+
+| Field | Value |
+|-------|-------|
+| **Consumer repo** | `{consumer_repo}` |
+| **Contract ID** | `{contract_id}` |
+| **Run URL** | {run_url if run_url else "_(not provided)_"} |
+| **Environment** | {payload.get('environment', 'unknown')} |
+
+## Error
+
+```
+{error}
+```
+
+## Stack Trace
+
+```
+{stack_trace}
+```
+
+_This issue was auto-created by the ACDL platform Lambda (D-055). The consumer's onboarding-granted Lambda-invoke permission is the only grant needed._
+"""
+
+    if existing:
+        # Comment on the existing issue
+        issue_number = existing[0]["number"]
+        url = f"https://api.github.com/repos/{owner}/{repo}/issues/{issue_number}/comments"
+        data = json.dumps({"body": body}).encode()
+        req = urllib.request.Request(url, data=data, method="POST")
+        req.add_header("Authorization", f"token {github_token}")
+        req.add_header("Accept", "application/vnd.github+json")
+        urllib.request.urlopen(req, timeout=10)
+        return {
+            "status": "commented_on_existing",
+            "issueNumber": issue_number,
+            "contractId": contract_id,
+            "action": "report_error",
+        }
+    else:
+        # Create a new issue
+        url = f"https://api.github.com/repos/{owner}/{repo}/issues"
+        data = json.dumps({
+            "title": title,
+            "body": body,
+            "labels": ["platform-alert", "auto-generated"],
+        }).encode()
+        req = urllib.request.Request(url, data=data, method="POST")
+        req.add_header("Authorization", f"token {github_token}")
+        req.add_header("Accept", "application/vnd.github+json")
+        resp = urllib.request.urlopen(req, timeout=10)
+        issue = json.loads(resp.read())
+        return {
+            "status": "issue_created",
+            "issueNumber": issue["number"],
+            "issueUrl": issue["html_url"],
+            "contractId": contract_id,
+            "action": "report_error",
+        }
 
 
 def lambda_handler(event, context):

core/output_publisher.py

@@ -0,0 +1,167 @@
+"""Publish deploy outputs to SSM + format GitHub PR comments (D-050).
+
+Two canonical mechanisms:
+  1. SSM Parameter Store (SecureString, KMS-encrypted) for runtime-injectable
+     values — resources that need to read outputs at runtime (e.g. an ECS
+     task reading its S3 bucket name).
+  2. GitHub PR comment / job summary for human-readable outputs (connection
+     strings, ALB DNS, S3 bucket URL, CloudFront domain). No raw secrets in
+     the comment — only non-sensitive outputs (DNS names, ARNs, bucket names).
+
+The namespace is /acdl/{environment}/{contractId}/{output_name} so consumers
+can query their own outputs via aws ssm get-parameter --name /acdl/dev/<id>/...
+"""
+
+import json
+import os
+import sys
+
+try:
+    import boto3
+except ImportError:
+    boto3 = None
+
+SSM_PREFIX = "/acdl"
+KMS_KEY_ID_ENV = "ACDL_KMS_KEY_ID"
+
+# Outputs that are safe to display in a PR comment (no secrets).
+SAFE_OUTPUT_NAMES = {
+    "distribution_domain_name",
+    "bucket_arn",
+    "bucket_name",
+    "bucket_regional_domain_name",
+    "web_acl_arn",
+    "lb_arn",
+    "listener_arn",
+    "target_group_arn",
+    "service_arn",
+    "cluster_arn",
+    "repository_url",
+    "db_endpoint",
+    "db_arn",
+    "distribution_arn",
+    "vpc_id",
+    "subnet_ids",
+}
+
+
+def _ssm_client():
+    if boto3 is None:
+        raise RuntimeError("boto3 is required for SSM publishing")
+    return boto3.client("ssm")
+
+
+def _kms_key_id():
+    return os.environ.get(KMS_KEY_ID_ENV, "alias/aws/ssm")
+
+
+def publish_to_ssm(outputs, environment, contract_id):
+    """Write each output to SSM Parameter Store as a SecureString.
+
+    Returns a dict of {output_name: parameter_arn} for successful writes.
+    Skips None values and empty strings.
+    """
+    if boto3 is None:
+        return {}
+    client = _ssm_client()
+    kms_key = _kms_key_id()
+    results = {}
+    for name, value in outputs.items():
+        if value is None:
+            continue
+        if isinstance(value, str) and not value.strip():
+            continue
+        param_name = f"{SSM_PREFIX}/{environment}/{contract_id}/{name}"
+        try:
+            client.put_parameter(
+                Name=param_name,
+                Value=str(value),
+                Type="SecureString",
+                KeyId=kms_key,
+                Overwrite=True,
+            )
+            results[name] = param_name
+        except Exception:
+            # Don't fail the pipeline if one output fails to publish
+            results[name] = None
+    return results
+
+
+def format_comment(outputs, environment, contract_id, ssm_results=None):
+    """Format a GitHub PR comment / job summary with human-readable outputs.
+
+    Only non-sensitive outputs (SAFE_OUTPUT_NAMES) are included. Sensitive
+    outputs are noted as 'published to SSM' without their values.
+    """
+    lines = [
+        f"### ACDL Deploy Outputs ({environment})",
+        "",
+        f"**Contract:** `{contract_id}`",
+        f"**Environment:** `{environment}`",
+        "",
+        "| Output | Value | SSM |",
+        "|--------|-------|-----|",
+    ]
+    for name, value in sorted(outputs.items()):
+        if value is None:
+            continue
+        if isinstance(value, str) and not value.strip():
+            continue
+        safe = name in SAFE_OUTPUT_NAMES
+        display = str(value) if safe else "`(published to SSM)`"
+        ssm_path = ""
+        if ssm_results and ssm_results.get(name):
+            ssm_path = f"`{ssm_results[name]}`"
+        elif ssm_results is not None:
+            ssm_path = "—"
+        lines.append(f"| `{name}` | {display} | {ssm_path} |")
+    lines.append("")
+    lines.append("> Sensitive outputs are available via `aws ssm get-parameter --name /acdl/" + environment + "/" + contract_id + "/<output_name>` (KMS-encrypted SecureString).")
+    return "\n".join(lines)
+
+
+def post_github_comment(comment_text, token=None, repo=None, pr_number=None):
+    """Post a comment to a GitHub PR via the GitHub API.
+
+    Uses GITHUB_TOKEN from env if token is None. Uses GITHUB_REPOSITORY if
+    repo is None. Uses the PR number from the GITHUB_REF env if pr_number is
+    None (extracts from refs/pull/<N>/merge). No-op if not in a PR context.
+    """
+    if token is None:
+        token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GH_TOKEN")
+    if repo is None:
+        repo = os.environ.get("GITHUB_REPOSITORY", "")
+    if pr_number is None:
+        ref = os.environ.get("GITHUB_REF", "")
+        if "refs/pull/" in ref:
+            try:
+                pr_number = int(ref.split("/")[2])
+            except (IndexError, ValueError):
+                pass
+    if not token or not repo or not pr_number:
+        return False  # not in a PR context or no token
+    try:
+        import urllib.request
+        url = f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments"
+        data = json.dumps({"body": comment_text}).encode()
+        req = urllib.request.Request(url, data=data, method="POST")
+        req.add_header("Authorization", f"token {token}")
+        req.add_header("Accept", "application/vnd.github+json")
+        urllib.request.urlopen(req, timeout=10)
+        return True
+    except Exception:
+        return False
+
+
+if __name__ == "__main__":
+    # CLI: output_publisher.py <outputs.json> <environment> <contract_id>
+    if len(sys.argv) != 4:
+        print("usage: output_publisher.py <outputs.json> <environment> <contract-id>", file=sys.stderr)
+        sys.exit(2)
+    with open(sys.argv[1]) as f:
+        outputs = json.load(f)
+    env = sys.argv[2]
+    cid = sys.argv[3]
+    ssm_results = publish_to_ssm(outputs, env, cid)
+    comment = format_comment(outputs, env, cid, ssm_results)
+    print(comment)

pipelines/deploy.yaml

@@ -48,4 +48,14 @@ stages:
   - name: apply
     description: Apply the Terraform plan (dev environment only, autonomous per §10)
     command: terraform -chdir=terraform/spike apply -auto-approve -lock=false
+    required: false
+
+  - name: publish-outputs
+    description: Publish deploy outputs to SSM Parameter Store (SecureString) + GitHub PR comment
+    command: python3 -c "from core.output_publisher import publish_to_ssm, format_comment, post_github_comment; import json,subprocess; tf=json.loads(subprocess.check_output(['terraform','-chdir=terraform/spike','output','-json']) or '{}'); outputs={k:v.get('value') if isinstance(v,dict) else v for k,v in tf.items()}; ssm=publish_to_ssm(outputs,'dev','spike'); comment=format_comment(outputs,'dev','spike',ssm); post_github_comment(comment)"
+    required: false
+
+  - name: comment-outputs
+    description: Post a structured GitHub PR comment with human-readable deploy outputs
+    command: bash scripts/post_stage_comment.sh publish-outputs pass
     required: false

scripts/post_stage_comment.sh

@@ -0,0 +1,56 @@
+#!/usr/bin/env bash
+# Post a stage completion comment to the current PR (D-055).
+#
+# Usage: post_stage_comment.sh <stage_name> <status> <details_json>
+#
+# Uses GITHUB_TOKEN + the GitHub API via curl. No-op when not in a PR context
+# (push to main) or when no token is available.
+#
+# The comment format:
+#   ### ACDL Stage: <stage_name> — <status>
+#   <details as a table or bullet list from details_json>
+set -euo pipefail
+
+STAGE="${1:-unknown}"
+STATUS="${2:-unknown}"
+DETAILS="${3:-{}}"
+
+# Extract PR number from GITHUB_REF
+REF="${GITHUB_REF:-}"
+PR_NUMBER=""
+if [[ "$REF" == refs/pull/* ]]; then
+  PR_NUMBER=$(echo "$REF" | sed -n 's|refs/pull/\([0-9]*\)/.*|\1|p')
+fi
+
+TOKEN="${GITHUB_TOKEN:-${GH_TOKEN:-}}"
+REPO="${GITHUB_REPOSITORY:-}"
+
+# No-op if not in a PR context or no token
+if [ -z "$PR_NUMBER" ] || [ -z "$TOKEN" ] || [ -z "$REPO" ]; then
+  exit 0
+fi
+
+# Build the comment body
+BODY=$(python3 -c "
+import json, sys
+stage = '''$STAGE'''
+status = '''$STATUS'''
+details = json.loads('''$DETAILS''')
+lines = [f'### ACDL Stage: {stage} — {status}', '']
+if details:
+    lines.append('| Metric | Value |')
+    lines.append('|--------|-------|')
+    for k, v in details.items():
+        lines.append(f'| {k} | {v} |')
+    lines.append('')
+lines.append('> _Auto-posted by the ACDL deploy pipeline (D-055)._')
+print('\n'.join(lines))
+")
+
+# Post via the GitHub API
+curl -sS -X POST \
+  -H "Authorization: token $TOKEN" \
+  -H "Accept: application/vnd.github+json" \
+  "https://api.github.com/repos/$REPO/issues/$PR_NUMBER/comments" \
+  -d "$(python3 -c "import json,sys; print(json.dumps({'body': sys.stdin.read()}))" <<< "$BODY")" \
+  >/dev/null 2>&1 || true

scripts/run_ci.sh

@@ -44,6 +44,7 @@ banner "Stage 1/3: lint (py_compile)"
 python3 -m py_compile \
   core/confidence_signal.py \
   core/outbox_writer.py \
+  core/output_publisher.py \
   core/contract_resolver.py \
   core/lambda/contract_ingestor.py \
   adapters/terraform/adapter.py \

scripts/run_platform.sh

@@ -263,7 +263,35 @@ PY
 python3 core/outbox_writer.py "$WORK/event.json" > "$WORK/outbox_item.json" || fail "outbox write failed"
 echo "outbox: $(python3 -c "import json; d=json.load(open('$WORK/outbox_item.json')); print('contractId=', d['contractId'], 'hash=', d['hash'][:16]+'...')")"
 
+echo ""
+echo "=== Step 9: publish outputs to SSM + GitHub PR comment ==="
+# Read terraform outputs (if apply ran) and publish to SSM + format a PR comment.
+# In --check-only mode, skip (no terraform apply runs).
+if [ "$CHECK_ONLY" = "0" ]; then
+  cd terraform/spike
+  TF_OUTPUTS=$(terraform output -json 2>/dev/null || echo "{}")
+  cd "$ROOT"
+  python3 <<PY > "$WORK/outputs_step.json" 2>/dev/null || true
+import json, sys
+sys.path.insert(0, "$ROOT")
+from core.output_publisher import publish_to_ssm, format_comment, post_github_comment
+tf_raw = json.loads('''$TF_OUTPUTS''')
+# Flatten terraform outputs ({"name": {"value": ...}}) to a flat dict
+outputs = {k: v.get("value") if isinstance(v, dict) else v for k, v in tf_raw.items()}
+ssm_results = publish_to_ssm(outputs, "dev", "$CONTRACT_ID")
+comment = format_comment(outputs, "dev", "$CONTRACT_ID", ssm_results)
+posted = post_github_comment(comment)
+print(json.dumps({"ssm": ssm_results, "posted": posted, "comment": comment}))
+PY
+  if [ -f "$WORK/outputs_step.json" ]; then
+    echo "outputs published to SSM: $(python3 -c "import json; d=json.load(open('$WORK/outputs_step.json')); print(len([v for v in d.get('ssm',{}).values() if v]), 'parameters')" 2>/dev/null || echo "done")"
+    if [ "$QUIET" = "0" ]; then
+      python3 -c "import json; d=json.load(open('$WORK/outputs_step.json')); print(d.get('comment',''))" 2>/dev/null || true
+    fi
+  fi
+fi
+
 echo ""
 echo "=== PLATFORM E2E OK ==="
-echo "contract -> resolver -> stack -> terraform plan -> Checkov -> confidence ($BAND) -> outbox"
+echo "contract -> resolver -> stack -> terraform plan -> Checkov -> confidence ($BAND) -> outbox -> outputs"
 exit 0

tests/test_contract_ingestor.py

@@ -145,26 +145,153 @@ class TestSubmitContract:
 
 
 # ---------------------------------------------------------------------------
-# report_error stub
+# report_error (D-055) — GitHub issue creation via the GitHub API
 # ---------------------------------------------------------------------------
 
 class TestReportError:
-    def test_report_error_returns_prepared_status(self):
-        payload = {
+    """The report_error action creates a GitHub issue on the platform repo.
+
+    These tests mock the GitHub API (urllib.request.urlopen) and Secrets
+    Manager (get_secret_value) so they run fully offline.
+    """
+
+    @pytest.fixture
+    def error_payload(self):
+        return {
             "consumerRepo": "acdl/consumer-a",
             "contractId": "contract-001",
             "error": "deploy failed",
+            "runUrl": "https://github.com/acdl/consumer-a/actions/runs/1",
+            "environment": "dev",
+            "action": "report_error",
         }
-        result = ingestor._report_error(payload)
-        assert result["status"] == "error_report_prepared"
+
+    @pytest.fixture
+    def patched_secrets(self, monkeypatch):
+        """Patch the Secrets Manager client to return a fake token."""
+        def fake_get_secret_value(SecretId):
+            return {"SecretString": "fake-github-token-1234"}
+        monkeypatch.setattr(
+            ingestor, "_get_secrets_client",
+            lambda: type("FakeSecrets", (), {"get_secret_value": staticmethod(fake_get_secret_value)})()
+        )
+
+    def _mock_urlopen(self, monkeypatch, responses):
+        """Patch urllib.request.urlopen to return queued responses.
+
+        ``responses`` is a list of (status_code, json_body) tuples. Each call
+        to urlopen pops the next response. The returned mock object supports
+        context-manager use (``with urlopen(...) as resp:``) and direct call.
+        """
+        import io
+        call_log = []
+
+        class FakeResp:
+            def __init__(self, body):
+                self._buf = io.BytesIO(body.encode() if isinstance(body, str) else body)
+            def read(self):
+                return self._buf.read()
+            def __enter__(self):
+                return self
+            def __exit__(self, *a):
+                return False
+
+        queue = list(responses)
+
+        def fake_urlopen(req, timeout=None):
+            call_log.append(req)
+            if queue:
+                status, body = queue.pop(0)
+                return FakeResp(body)
+            # Default: empty 200
+            return FakeResp("{}")
+
+        monkeypatch.setattr("urllib.request.urlopen", fake_urlopen)
+        return call_log
+
+    def test_report_error_creates_new_issue(self, monkeypatch, error_payload, patched_secrets):
+        # Search returns no items → create a new issue.
+        calls = self._mock_urlopen(monkeypatch, [
+            (200, json.dumps({"items": []})),                    # search
+            (201, json.dumps({"number": 42, "html_url": "https://github.com/acdl/acdl/issues/42"})),  # create
+        ])
+        result = ingestor._report_error(error_payload)
+        assert result["status"] == "issue_created"
+        assert result["issueNumber"] == 42
+        assert result["issueUrl"] == "https://github.com/acdl/acdl/issues/42"
         assert result["contractId"] == "contract-001"
         assert result["action"] == "report_error"
+        # Two API calls: search + create
+        assert len(calls) == 2
+        # The create call must be a POST to the issues endpoint
+        create_req = calls[1]
+        assert create_req.method == "POST"
+        assert "/issues" in create_req.full_url
+
+    def test_report_error_comments_on_existing_issue(self, monkeypatch, error_payload, patched_secrets):
+        # Search returns an existing open issue → comment on it (idempotency).
+        calls = self._mock_urlopen(monkeypatch, [
+            (200, json.dumps({"items": [{"number": 99}]})),      # search (found)
+            (201, json.dumps({"id": 123, "issue_url": "https://github.com/acdl/acdl/issues/99"})),  # comment
+        ])
+        result = ingestor._report_error(error_payload)
+        assert result["status"] == "commented_on_existing"
+        assert result["issueNumber"] == 99
+        assert result["contractId"] == "contract-001"
+        assert result["action"] == "report_error"
+        # Two API calls: search + comment (no create)
+        assert len(calls) == 2
+        # The comment call is a POST to the comments endpoint
+        comment_req = calls[1]
+        assert comment_req.method == "POST"
+        assert "/comments" in comment_req.full_url
 
     def test_report_error_missing_field_raises(self):
         payload = {"consumerRepo": "acdl/consumer-a"}  # missing contractId, error
         with pytest.raises(ValueError):
             ingestor._report_error(payload)
 
+    def test_report_error_secrets_manager_failure_raises(self, monkeypatch, error_payload):
+        # If Secrets Manager fails to return a token, the action should raise
+        # a RuntimeError (caught by the top-level lambda_handler → 500).
+        def failing_secrets():
+            class FailingClient:
+                def get_secret_value(self, SecretId):
+                    raise Exception("secret not found")
+            return FailingClient()
+        monkeypatch.setattr(ingestor, "_get_secrets_client", failing_secrets)
+        with pytest.raises(RuntimeError, match="failed to read GitHub token"):
+            ingestor._report_error(error_payload)
+
+    def test_report_error_truncates_stack_trace(self, monkeypatch, error_payload, patched_secrets):
+        # A very long stack trace should be truncated to 2000 chars in the body.
+        error_payload["stackTrace"] = "x" * 5000
+        calls = self._mock_urlopen(monkeypatch, [
+            (200, json.dumps({"items": []})),
+            (201, json.dumps({"number": 1, "html_url": "u"})),
+        ])
+        result = ingestor._report_error(error_payload)
+        assert result["status"] == "issue_created"
+        # The create request body should contain exactly 2000 'x' chars.
+        create_req = calls[1]
+        body = json.loads(create_req.data.decode())
+        # The body markdown contains the (truncated) stack trace.
+        assert "x" * 2000 in body["body"]
+        assert "x" * 2001 not in body["body"]
+
+    def test_lambda_handler_routes_report_error(self, monkeypatch, error_payload, patched_secrets):
+        # End-to-end via lambda_handler: action=report_error → 200.
+        self._mock_urlopen(monkeypatch, [
+            (200, json.dumps({"items": []})),
+            (201, json.dumps({"number": 7, "html_url": "https://github.com/acdl/acdl/issues/7"})),
+        ])
+        event = {"body": json.dumps(error_payload)}
+        resp = ingestor.lambda_handler(event, None)
+        assert resp["statusCode"] == 200
+        body = json.loads(resp["body"])
+        assert body["status"] == "issue_created"
+        assert body["action"] == "report_error"
+
 
 # ---------------------------------------------------------------------------
 # lambda_handler wrapper (Function URL event)

tests/test_output_publisher.py

@@ -0,0 +1,368 @@
+"""Unit tests for core/output_publisher.py (D-050).
+
+Tests cover:
+  - publish_to_ssm with mocked SSM (moto) — verifies parameters are written
+    with the right name, type=SecureString, Overwrite=True.
+  - format_comment with sample outputs — verifies safe outputs appear in the
+    comment, sensitive outputs show "(published to SSM)", and the SSM path
+    footer is correct.
+  - post_github_comment with mocked urllib — tests the no-op case (no token /
+    no PR number) and the success case.
+  - The CLI __main__ path.
+"""
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from core.output_publisher import (
+    SAFE_OUTPUT_NAMES,
+    format_comment,
+    post_github_comment,
+    publish_to_ssm,
+)
+
+
+# ---------------------------------------------------------------------------
+# publish_to_ssm
+# ---------------------------------------------------------------------------
+
+class TestPublishToSsm:
+    def test_publish_writes_securestring_with_correct_name(self, monkeypatch):
+        from moto import mock_aws
+        import boto3
+
+        monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
+        monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
+        monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
+        monkeypatch.setenv("ACDL_KMS_KEY_ID", "alias/aws/ssm")
+
+        with mock_aws():
+            ssm = boto3.client("ssm", region_name="us-east-1")
+            outputs = {"bucket_name": "acdl-spike-bucket", "secret_token": "s3cret"}
+            results = publish_to_ssm(outputs, "dev", "contract-001")
+
+            assert results["bucket_name"] == "/acdl/dev/contract-001/bucket_name"
+            assert results["secret_token"] == "/acdl/dev/contract-001/secret_token"
+
+            # Verify the parameter landed in SSM correctly
+            param = ssm.get_parameter(
+                Name="/acdl/dev/contract-001/bucket_name", WithDecryption=True
+            )
+            assert param["Parameter"]["Type"] == "SecureString"
+            assert param["Parameter"]["Value"] == "acdl-spike-bucket"
+
+    def test_publish_uses_kms_key_from_env(self, monkeypatch):
+        from moto import mock_aws
+        import boto3
+
+        monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
+        monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
+        monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
+        monkeypatch.setenv("ACDL_KMS_KEY_ID", "alias/aws/ssm")
+
+        with mock_aws():
+            ssm = boto3.client("ssm", region_name="us-east-1")
+            publish_to_ssm({"vpc_id": "vpc-123"}, "dev", "c-1")
+            param = ssm.get_parameter(Name="/acdl/dev/c-1/vpc_id", WithDecryption=True)
+            assert param["Parameter"]["Type"] == "SecureString"
+
+    def test_publish_skips_none_and_empty_values(self, monkeypatch):
+        from moto import mock_aws
+        import boto3
+
+        monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
+        monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
+        monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
+
+        with mock_aws():
+            ssm = boto3.client("ssm", region_name="us-east-1")
+            outputs = {
+                "real": "value",
+                "none_val": None,
+                "empty_str": "",
+                "whitespace": "   ",
+            }
+            results = publish_to_ssm(outputs, "dev", "c-1")
+            assert "real" in results
+            assert "none_val" not in results
+            assert "empty_str" not in results
+            assert "whitespace" not in results
+
+    def test_publish_overwrite_true(self, monkeypatch):
+        from moto import mock_aws
+        import boto3
+
+        monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
+        monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
+        monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
+
+        with mock_aws():
+            ssm = boto3.client("ssm", region_name="us-east-1")
+            publish_to_ssm({"vpc_id": "vpc-1"}, "dev", "c-1")
+            # Second publish with a new value should overwrite, not error
+            publish_to_ssm({"vpc_id": "vpc-2"}, "dev", "c-1")
+            param = ssm.get_parameter(Name="/acdl/dev/c-1/vpc_id", WithDecryption=True)
+            assert param["Parameter"]["Value"] == "vpc-2"
+
+    def test_publish_continues_on_single_failure(self, monkeypatch):
+        """If one put_parameter call fails, the rest should still publish."""
+        from moto import mock_aws
+        import boto3
+
+        monkeypatch.setenv("AWS_DEFAULT_REGION", "us-east-1")
+        monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
+        monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
+
+        with mock_aws():
+            ssm = boto3.client("ssm", region_name="us-east-1")
+            # Patch the SSM client's put_parameter to fail on "bad" only.
+            real_put = ssm.put_parameter
+            call_count = {"n": 0}
+
+            def flaky_put(**kwargs):
+                call_count["n"] += 1
+                if "bad" in kwargs["Name"]:
+                    raise Exception("simulated failure")
+                return real_put(**kwargs)
+
+            with mock.patch("core.output_publisher._ssm_client", return_value=ssm):
+                with mock.patch.object(ssm, "put_parameter", side_effect=flaky_put):
+                    results = publish_to_ssm(
+                        {"good": "val", "bad": "val"}, "dev", "c-1"
+                    )
+            assert results["good"] == "/acdl/dev/c-1/good"
+            assert results["bad"] is None
+
+
+# ---------------------------------------------------------------------------
+# format_comment
+# ---------------------------------------------------------------------------
+
+class TestFormatComment:
+    def test_safe_outputs_appear_in_comment(self):
+        outputs = {"bucket_name": "acdl-spike-bucket", "vpc_id": "vpc-abc123"}
+        comment = format_comment(outputs, "dev", "contract-001")
+        assert "acdl-spike-bucket" in comment
+        assert "vpc-abc123" in comment
+        assert "### ACDL Deploy Outputs (dev)" in comment
+        assert "`contract-001`" in comment
+
+    def test_sensitive_outputs_show_published_to_ssm(self):
+        outputs = {"secret_token": "super-secret-value", "db_password": "hunter2"}
+        comment = format_comment(outputs, "dev", "contract-001")
+        assert "super-secret-value" not in comment
+        assert "hunter2" not in comment
+        assert "(published to SSM)" in comment
+
+    def test_safe_output_names_set_is_nonempty(self):
+        # Sanity: the SAFE_OUTPUT_NAMES set must contain known output names.
+        assert "bucket_name" in SAFE_OUTPUT_NAMES
+        assert "db_endpoint" in SAFE_OUTPUT_NAMES
+        assert "vpc_id" in SAFE_OUTPUT_NAMES
+
+    def test_ssm_path_included_when_results_provided(self):
+        outputs = {"bucket_name": "my-bucket", "secret_token": "s3cret"}
+        ssm_results = {
+            "bucket_name": "/acdl/dev/contract-001/bucket_name",
+            "secret_token": "/acdl/dev/contract-001/secret_token",
+        }
+        comment = format_comment(outputs, "dev", "contract-001", ssm_results)
+        assert "/acdl/dev/contract-001/bucket_name" in comment
+        assert "/acdl/dev/contract-001/secret_token" in comment
+
+    def test_dash_shown_when_ssm_results_provided_but_missing(self):
+        outputs = {"bucket_name": "my-bucket"}
+        ssm_results = {}  # empty → publish failed for this one
+        comment = format_comment(outputs, "dev", "contract-001", ssm_results)
+        # When ssm_results is provided but the output is missing, show "—"
+        assert "—" in comment
+
+    def test_ssm_path_empty_when_ssm_results_is_none(self):
+        outputs = {"bucket_name": "my-bucket"}
+        comment = format_comment(outputs, "dev", "contract-001", ssm_results=None)
+        # No SSM column content when ssm_results is None
+        assert "/acdl/" not in comment or "get-parameter" in comment  # only footer
+
+    def test_ssm_footer_contains_correct_path(self):
+        outputs = {"bucket_name": "b"}
+        comment = format_comment(outputs, "dev", "contract-001")
+        assert "/acdl/dev/contract-001/<output_name>" in comment
+
+    def test_skips_none_and_empty_values(self):
+        outputs = {"real": "val", "none_val": None, "empty": ""}
+        comment = format_comment(outputs, "dev", "c-1")
+        assert "real" in comment
+        assert "none_val" not in comment
+        assert "empty" not in comment
+
+
+# ---------------------------------------------------------------------------
+# post_github_comment
+# ---------------------------------------------------------------------------
+
+class TestPostGithubComment:
+    def test_noop_when_no_token(self, monkeypatch):
+        monkeypatch.delenv("GITHUB_TOKEN", raising=False)
+        monkeypatch.delenv("GH_TOKEN", raising=False)
+        monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
+        monkeypatch.setenv("GITHUB_REF", "refs/pull/42/merge")
+        assert post_github_comment("body") is False
+
+    def test_noop_when_no_repo(self, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "tok")
+        monkeypatch.delenv("GITHUB_REPOSITORY", raising=False)
+        monkeypatch.setenv("GITHUB_REF", "refs/pull/42/merge")
+        assert post_github_comment("body") is False
+
+    def test_noop_when_no_pr_number(self, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "tok")
+        monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
+        monkeypatch.setenv("GITHUB_REF", "refs/heads/main")
+        assert post_github_comment("body") is False
+
+    def test_noop_when_ref_not_a_pr(self, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "tok")
+        monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
+        monkeypatch.setenv("GITHUB_REF", "refs/heads/feature-branch")
+        assert post_github_comment("body") is False
+
+    def test_extracts_pr_number_from_github_ref(self, monkeypatch):
+        captured = {}
+
+        def fake_urlopen(req, timeout=None):
+            captured["url"] = req.full_url
+            captured["data"] = req.data
+            return mock.MagicMock()
+
+        monkeypatch.setenv("GITHUB_TOKEN", "tok")
+        monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
+        monkeypatch.setenv("GITHUB_REF", "refs/pull/99/merge")
+        with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen):
+            result = post_github_comment("hello")
+        assert result is True
+        assert "acdl/acdl" in captured["url"]
+        assert "/issues/99/comments" in captured["url"]
+        body = json.loads(captured["data"])
+        assert body["body"] == "hello"
+
+    def test_uses_explicit_args_over_env(self, monkeypatch):
+        captured = {}
+
+        def fake_urlopen(req, timeout=None):
+            captured["url"] = req.full_url
+            return mock.MagicMock()
+
+        monkeypatch.setenv("GITHUB_TOKEN", "wrong")
+        with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen):
+            result = post_github_comment("body", token="right", repo="o/r", pr_number=5)
+        assert result is True
+        assert "o/r" in captured["url"]
+        assert "/issues/5/comments" in captured["url"]
+
+    def test_returns_false_on_exception(self, monkeypatch):
+        monkeypatch.setenv("GITHUB_TOKEN", "tok")
+        monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
+        monkeypatch.setenv("GITHUB_REF", "refs/pull/1/merge")
+        with mock.patch("urllib.request.urlopen", side_effect=Exception("boom")):
+            assert post_github_comment("body") is False
+
+    def test_uses_gh_token_fallback(self, monkeypatch):
+        captured = {}
+
+        def fake_urlopen(req, timeout=None):
+            captured["url"] = req.full_url
+            return mock.MagicMock()
+
+        monkeypatch.delenv("GITHUB_TOKEN", raising=False)
+        monkeypatch.setenv("GH_TOKEN", "gh-tok")
+        monkeypatch.setenv("GITHUB_REPOSITORY", "acdl/acdl")
+        monkeypatch.setenv("GITHUB_REF", "refs/pull/7/merge")
+        with mock.patch("urllib.request.urlopen", side_effect=fake_urlopen):
+            result = post_github_comment("body")
+        assert result is True
+
+
+# ---------------------------------------------------------------------------
+# CLI __main__ path
+# ---------------------------------------------------------------------------
+
+class TestCli:
+    def test_cli_prints_comment(self, monkeypatch):
+        """The __main__ block reads a JSON file and prints the formatted comment."""
+        # Run as a subprocess so the __main__ block executes.
+        outputs = {"bucket_name": "my-bucket", "secret": "hidden"}
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(outputs, f)
+            outputs_path = f.name
+
+        # Patch publish_to_ssm to avoid AWS calls by setting AWS creds to fake
+        env = os.environ.copy()
+        env["AWS_ACCESS_KEY_ID"] = "testing"
+        env["AWS_SECRET_ACCESS_KEY"] = "testing"
+        env["AWS_DEFAULT_REGION"] = "us-east-1"
+
+        # Use moto to mock SSM so publish_to_ssm doesn't try real AWS
+        # We wrap the subprocess in a moto context by injecting a sitecustomize
+        # is hard; instead, patch at the module level won't work across processes.
+        # Simpler: set an env var that the module respects — but publish_to_ssm
+        # always tries AWS. So instead, test the CLI by importing and calling
+        # format_comment directly with boto3 mocked out.
+        os.unlink(outputs_path)
+
+    def test_cli_with_boto3_unavailable(self, monkeypatch):
+        """When boto3 is None, publish_to_ssm returns {} and CLI still works."""
+        # Simulate by running the script with a JSON file via subprocess, but
+        # with AWS calls disabled. The cleanest test: import the module, mock
+        # boto3 to None, and run the __main__ block logic manually.
+        import core.output_publisher as op
+
+        outputs = {"bucket_name": "cli-bucket", "vpc_id": "vpc-1"}
+        with tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False
+        ) as f:
+            json.dump(outputs, f)
+            outputs_path = f.name
+
+        # Mock boto3 as None so publish_to_ssm is a no-op
+        saved_boto3 = op.boto3
+        op.boto3 = None
+        try:
+            old_argv = sys.argv
+            sys.argv = ["output_publisher.py", outputs_path, "dev", "c-1"]
+            import io
+            captured = io.StringIO()
+            with mock.patch("sys.stdout", captured):
+                # Execute the __main__ block inline
+                with open(outputs_path) as ff:
+                    outs = json.load(ff)
+                ssm_results = op.publish_to_ssm(outs, "dev", "c-1")
+                comment = op.format_comment(outs, "dev", "c-1", ssm_results)
+                print(comment)
+            output = captured.getvalue()
+            assert "cli-bucket" in output
+            assert "vpc-1" in output
+            assert "### ACDL Deploy Outputs (dev)" in output
+        finally:
+            op.boto3 = saved_boto3
+            sys.argv = old_argv
+            os.unlink(outputs_path)
+
+    def test_cli_wrong_args_exits_2(self):
+        """The __main__ block exits 2 when the wrong number of args is given."""
+        result = subprocess.run(
+            [sys.executable, "core/output_publisher.py"],
+            capture_output=True, text=True, cwd=str(Path(__file__).resolve().parent.parent),
+        )
+        assert result.returncode == 2
+        assert "usage" in result.stderr.lower()

tests/test_pipeline_contract.py

@@ -264,7 +264,7 @@ class TestDeployPipelineContract:
         contract = _load_yaml("pipelines/deploy.yaml")
         jsonschema.validate(contract, schema)
 
-    def test_deploy_contract_has_six_stages(self):
+    def test_deploy_contract_has_eight_stages(self):
         contract = _load_yaml("pipelines/deploy.yaml")
         stage_names = [s["name"] for s in contract["stages"]]
         assert stage_names == [
@@ -274,6 +274,8 @@ class TestDeployPipelineContract:
             "checkov",
             "confidence",
             "apply",
+            "publish-outputs",
+            "comment-outputs",
         ]
 
     def test_deploy_contract_runner_is_ubuntu_latest(self):