Merge pull request #472 from QuantStrategyLab/codex/reconciliation-request-correlation-20260905

fix: correlate IBKR reconciliation receipts

Author: Pigbibi

.github/workflows/collect-reconciliation-evidence.yml

@@ -93,6 +93,7 @@ jobs:
           set -euo pipefail
           test -n "$SERVICE_URL"
           job_name="ibkr-reconcile-${PROFILE}-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          request_id="$(python -c 'import uuid; print(uuid.uuid4())')"
           requested_at="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
           case "$PROFILE" in
             soxl_soxx_trend_income) delay_minutes=3 ;;
@@ -116,6 +117,7 @@ jobs:
             --time-zone 'Etc/UTC' \
             --uri "${SERVICE_URL}/reconcile" \
             --http-method POST \
+            --headers "X-QSL-Reconciliation-Request-Id=${request_id}" \
             --oidc-service-account-email "$GCP_SCHEDULER_SERVICE_ACCOUNT" \
             --oidc-token-audience "$SERVICE_URL" \
             --attempt-deadline '300s' \
@@ -124,6 +126,7 @@ jobs:
             --quiet >/dev/null
           {
             echo "job_name=$job_name"
+            echo "request_id=$request_id"
             echo "requested_at=$requested_at"
           } >> "$GITHUB_OUTPUT"
 
@@ -135,19 +138,19 @@ jobs:
         env:
           PROFILE: ${{ matrix.profile }}
           SERVICE: ${{ matrix.service }}
-          SERVICE_URL: ${{ steps.audience.outputs.service_url }}
+          REQUEST_ID: ${{ steps.scheduler.outputs.request_id }}
           REQUESTED_AT: ${{ steps.scheduler.outputs.requested_at }}
         run: |
           set -euo pipefail
-          test -n "$SERVICE_URL"
+          test -n "$REQUEST_ID"
           test -n "$REQUESTED_AT"
           mkdir -p reports
           log_entry=''
           logging_read_failures=0
-          for attempt in $(seq 1 96); do
+          for _ in $(seq 1 96); do
             logging_output=''
             logging_read_status=0
-            logging_output="$(gcloud logging read "resource.type=\"cloud_run_revision\" AND resource.labels.service_name=\"${SERVICE}\" AND timestamp>=\"${REQUESTED_AT}\" AND textPayload:\"execution_report gs://\"" --project "$GCP_PROJECT_ID" --freshness=15m --limit=10 --format=json 2>/dev/null)" || logging_read_status=$?
+            logging_output="$(gcloud logging read "resource.type=\"cloud_run_revision\" AND resource.labels.service_name=\"${SERVICE}\" AND timestamp>=\"${REQUESTED_AT}\" AND textPayload:\"reconciliation_receipt_ready request_id=${REQUEST_ID} report_uri=gs://\"" --project "$GCP_PROJECT_ID" --freshness=15m --limit=10 --format=json 2>/dev/null)" || logging_read_status=$?
             if [ "$logging_read_status" -ne 0 ]; then
               logging_read_failures=$((logging_read_failures + 1))
               sleep 5
@@ -157,7 +160,7 @@ jobs:
               echo "reconciliation_log_query_failed class=invalid_logging_json" >&2
               exit 1
             fi
-            log_entry="$(jq -c 'first(.[] | select(.textPayload? | startswith("execution_report gs://")) | select(.resource.labels.revision_name? | type == "string" and length > 0)) // empty' <<< "$logging_output")"
+            log_entry="$(jq -c --arg request_id "$REQUEST_ID" 'first(.[] | select(.textPayload? | type == "string") | select(.textPayload | startswith("reconciliation_receipt_ready request_id=" + $request_id + " report_uri=gs://")) | select(.resource.labels.revision_name? | type == "string" and length > 0)) // empty' <<< "$logging_output")"
             if [ -n "$log_entry" ]; then
               break
             fi
@@ -173,16 +176,18 @@ jobs:
             fi
             exit 1
           fi
-          report_uri="$(jq -er '.textPayload | capture("^execution_report (?<uri>gs://[^[:space:]]+)$").uri' <<< "$log_entry")"
+          receipt="$(jq -cer --arg request_id "$REQUEST_ID" '.textPayload | capture("^reconciliation_receipt_ready request_id=(?<request_id>[0-9a-f-]{36}) report_uri=(?<uri>gs://[^[:space:]]+)$") | select(.request_id == $request_id)' <<< "$log_entry")"
+          report_uri="$(jq -er '.uri' <<< "$receipt")"
           serving_revision="$(jq -er '.resource.labels.revision_name | select(type == "string" and length > 0)' <<< "$log_entry")"
           revision_labels="$(gcloud run revisions describe "$serving_revision" --project "$GCP_PROJECT_ID" --region "$GCP_REGION" --format='json(metadata.labels)')"
           service_revision_commit_sha="$(jq -er '."commit-sha" | select(type == "string" and test("^[0-9a-f]{40}$"))' <<< "$revision_labels")"
           service_deploy_run_id="$(jq -er '."github-run-id" | select(type == "string" and test("^[0-9]+$"))' <<< "$revision_labels")"
-          gcloud storage cat "$report_uri" | jq --arg profile "$PROFILE" '
+          gcloud storage cat "$report_uri" | jq --arg profile "$PROFILE" --arg request_id "$REQUEST_ID" '
             .diagnostics.broker_reconciliation as $candidate
             | {
                 schema_version: "ibkr_reconciliation_artifact.v1",
                 strategy_profile: $profile,
+                reconciliation_request_id: .diagnostics.reconciliation_request_id,
                 report_status: .status,
                 reconciliation: $candidate,
                 errors: [
@@ -194,9 +199,10 @@ jobs:
                 ]
               }
           ' > "reports/${PROFILE}.json"
-          jq -e --arg profile "$PROFILE" '
+          jq -e --arg profile "$PROFILE" --arg request_id "$REQUEST_ID" '
             .schema_version == "ibkr_reconciliation_artifact.v1"
             and .strategy_profile == $profile
+            and .reconciliation_request_id == $request_id
             and .reconciliation.schema_version == "ibkr_reconciliation_candidate.v1"
             and .reconciliation.evidence.platform_id == "ibkr"
             and .reconciliation.evidence.strategy_profile == $profile

application/reconciliation_reporting.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping
+import re
 from typing import Any
 
 _CANDIDATE_FIELDS = (
@@ -52,6 +53,19 @@
     "started_at",
     "finished_at",
 )
+_RECONCILIATION_REQUEST_ID_PATTERN = re.compile(
+    r"[0-9a-f]{8}-(?:[0-9a-f]{4}-){3}[0-9a-f]{12}"
+)
+
+
+def normalize_reconciliation_request_id(value: object) -> str | None:
+    """Return a safe opaque correlation ID or reject the untrusted value."""
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip().lower()
+    if _RECONCILIATION_REQUEST_ID_PATTERN.fullmatch(normalized):
+        return normalized
+    return None
 
 
 def _selected_mapping(value: object, fields: tuple[str, ...]) -> dict[str, Any]:
@@ -89,6 +103,13 @@ def build_persistable_reconciliation_report(report: Mapping[str, object]) -> dic
     )
     if candidate:
         safe_report["diagnostics"]["broker_reconciliation"] = candidate
+    request_id = normalize_reconciliation_request_id(
+        (report.get("diagnostics") or {}).get("reconciliation_request_id")
+        if isinstance(report.get("diagnostics"), Mapping)
+        else None
+    )
+    if request_id:
+        safe_report["diagnostics"]["reconciliation_request_id"] = request_id
     failure = _selected_mapping(
         (report.get("diagnostics") or {}).get("broker_reconciliation_failure")
         if isinstance(report.get("diagnostics"), Mapping)

main.py

@@ -28,7 +28,10 @@
     build_reconciliation_candidate,
     collect_read_only_reconciliation_observations,
 )
-from application.reconciliation_reporting import build_persistable_reconciliation_report
+from application.reconciliation_reporting import (
+    build_persistable_reconciliation_report,
+    normalize_reconciliation_request_id,
+)
 from application.runtime_broker_adapters import (
     IBKRGatewayUnavailableError,
     IBKRTradingPermissionError,
@@ -1882,9 +1885,14 @@ def _handle_reconciliation():
     ib = None
     log_context = None
     report = None
+    reconciliation_request_id = normalize_reconciliation_request_id(
+        request.headers.get("X-QSL-Reconciliation-Request-Id")
+    )
     try:
         log_context = build_request_log_context()
         report = build_execution_report(log_context, dry_run_only_override=True)
+        if reconciliation_request_id is not None:
+            report.setdefault("diagnostics", {})["reconciliation_request_id"] = reconciliation_request_id
         runtime_target = RUNTIME_SETTINGS.runtime_target
         if runtime_target is None:
             raise IBKRReconciliationReadError(
@@ -1999,6 +2007,12 @@ def _handle_reconciliation():
                     and not any(character.isspace() for character in report_path)
                 ):
                     print(f"execution_report {report_path}", flush=True)
+                    if reconciliation_request_id is not None:
+                        print(
+                            "reconciliation_receipt_ready "
+                            f"request_id={reconciliation_request_id} report_uri={report_path}",
+                            flush=True,
+                        )
                 else:
                     print("broker reconciliation report persisted", flush=True)
         except Exception as persist_exc:

tests/test_reconciliation_evidence_workflow.py

@@ -82,3 +82,21 @@ def test_reconciliation_evidence_preserves_sanitized_log_query_failure_class() -
     assert 'reconciliation_log_query_timeout class=no_matching_candidate' in workflow
     assert 'gcloud logging read' in workflow
     assert '--format=json 2>/dev/null' in workflow
+
+
+def test_reconciliation_evidence_binds_scheduler_request_to_exact_receipt() -> None:
+    workflow = WORKFLOW.read_text(encoding="utf-8")
+
+    required_markers = (
+        "request_id=\"$(python -c 'import uuid; print(uuid.uuid4())')\"",
+        "X-QSL-Reconciliation-Request-Id=${request_id}",
+        "echo \"request_id=$request_id\"",
+        "REQUEST_ID: ${{ steps.scheduler.outputs.request_id }}",
+        "reconciliation_receipt_ready request_id=${REQUEST_ID} report_uri=gs://",
+        "--arg request_id \"$REQUEST_ID\"",
+        ".reconciliation_request_id == $request_id",
+    )
+    for marker in required_markers:
+        assert marker in workflow
+
+    assert 'textPayload:\"execution_report gs://\"' not in workflow

tests/test_reconciliation_reporting.py

@@ -74,6 +74,7 @@ def test_persistable_reconciliation_report_keeps_only_safe_evidence(tmp_path) ->
         },
         "diagnostics": {
             "broker_reconciliation": candidate,
+            "reconciliation_request_id": "853a2e08-9396-4fe8-89ee-59fb17e40a1d",
             "ib_gateway_host": marker,
         },
         "artifacts": {"strategy_config_path": marker},
@@ -93,6 +94,7 @@ def test_persistable_reconciliation_report_keeps_only_safe_evidence(tmp_path) ->
 
     assert result.local_path is not None
     assert payload["diagnostics"]["broker_reconciliation"] == candidate
+    assert payload["diagnostics"]["reconciliation_request_id"] == "853a2e08-9396-4fe8-89ee-59fb17e40a1d"
     assert payload["errors"] == [
         {
             "stage": "broker_reconciliation",
@@ -111,3 +113,25 @@ def test_persistable_reconciliation_report_keeps_only_safe_evidence(tmp_path) ->
     assert "runtime_target" not in payload
     assert "account_scope" not in payload
     assert "service_name" not in payload
+
+
+def test_persistable_reconciliation_report_rejects_untrusted_request_id() -> None:
+    report = {
+        "schema_version": "runtime_report.v1",
+        "platform": "ibkr",
+        "deploy_target": "cloud_run",
+        "strategy_profile": "global_etf_rotation",
+        "run_id": "run-001",
+        "run_source": "cloud_run",
+        "status": "ok",
+        "started_at": "2026-09-04T00:00:00Z",
+        "finished_at": "2026-09-04T00:01:00Z",
+        "summary": {},
+        "diagnostics": {"reconciliation_request_id": "unsafe value\nreport_uri=gs://x"},
+        "artifacts": {},
+        "errors": [],
+    }
+
+    sanitized = build_persistable_reconciliation_report(report)
+
+    assert "reconciliation_request_id" not in sanitized["diagnostics"]

tests/test_request_handling.py

@@ -1446,7 +1446,11 @@ def persist_report(report, **_kwargs):
 
     monkeypatch.setattr(strategy_module, "persist_execution_report", persist_report)
 
-    with strategy_module.app.test_request_context("/reconcile", method="POST"):
+    with strategy_module.app.test_request_context(
+        "/reconcile",
+        method="POST",
+        headers={"X-QSL-Reconciliation-Request-Id": "853a2e08-9396-4fe8-89ee-59fb17e40a1d"},
+    ):
         body, status = strategy_module.handle_reconciliation()
 
     assert (body, status) == ("Error", 503)
@@ -1457,6 +1461,9 @@ def persist_report(report, **_kwargs):
             "failure_category": "broker_reconciliation",
         }
     ]
+    assert observed["report"]["diagnostics"]["reconciliation_request_id"] == (
+        "853a2e08-9396-4fe8-89ee-59fb17e40a1d"
+    )
     failed_event = observed["events"][-1]
     assert failed_event[0] == "broker_reconciliation_failed"
     assert "error_message" not in failed_event[1]
@@ -1465,7 +1472,12 @@ def persist_report(report, **_kwargs):
     assert "demo-account" not in serialized
     output = capsys.readouterr().out
     assert marker not in output
-    assert output == "execution_report gs://private-reports/reconciliation.json\n"
+    assert output == (
+        "execution_report gs://private-reports/reconciliation.json\n"
+        "reconciliation_receipt_ready "
+        "request_id=853a2e08-9396-4fe8-89ee-59fb17e40a1d "
+        "report_uri=gs://private-reports/reconciliation.json\n"
+    )
 
 
 def test_handle_reconciliation_hides_disconnect_and_persistence_error_details(