From a595b708204572d1169755f9fa38a41e9d9a5a2f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 10 Sep 2026 02:42:06 +0800 Subject: [PATCH] feat: consume eligible SOXL watcher diagnoses as bounded learning Co-Authored-By: Codex --- .../strategy_optimization_watcher.yml | 287 +++++++++++ scripts/run_soxl_manual_learning.py | 449 +++++++++++++++++- service/research_task.py | 64 ++- tests/test_run_soxl_manual_learning.py | 222 +++++++++ ..._strategy_optimization_watcher_workflow.py | 15 + tests/test_strategy_watch.py | 42 ++ 6 files changed, 1063 insertions(+), 16 deletions(-) diff --git a/.github/workflows/strategy_optimization_watcher.yml b/.github/workflows/strategy_optimization_watcher.yml index eed82391..c0d2298d 100644 --- a/.github/workflows/strategy_optimization_watcher.yml +++ b/.github/workflows/strategy_optimization_watcher.yml @@ -46,6 +46,9 @@ jobs: strategy-optimization-watcher: runs-on: ubuntu-latest timeout-minutes: 15 + outputs: + soxl_learning_ready: ${{ steps.soxl-watcher-learning-ready.outputs.ready }} + soxl_p1_manifest_sha256: ${{ steps.soxl-watcher-learning-ready.outputs.p1_manifest_sha256 }} env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" SOURCE_REPO: ${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || 'QuantStrategyLab/CryptoLivePoolPipelines' }} @@ -285,8 +288,10 @@ jobs: python -m scripts.run_strategy_optimization_watcher | tee data/output/strategy_optimization_watcher/result.json - name: Run one bounded Codex-only research diagnosis + id: research-diagnosis if: steps.fetch-metrics.outputs.downloaded == 'true' && github.event_name == 'schedule' env: + GH_TOKEN: ${{ steps.source_app_token.outputs.token || github.token }} CODEX_AUDIT_SERVICE_URL: ${{ secrets.CODEX_AUDIT_SERVICE_URL }} CODEX_AUDIT_SERVICE_AUDIENCE: ${{ vars.CODEX_AUDIT_SERVICE_AUDIENCE || 'quant-codex-audit' }} RESEARCH_DIAGNOSIS_MAX_PER_RUN: "1" @@ -298,6 +303,33 @@ jobs: --max-per-run "${RESEARCH_DIAGNOSIS_MAX_PER_RUN}" \ | tee data/output/strategy_optimization_watcher/research-diagnosis.json + - name: Resolve exact SOXL watcher learning handoff + id: soxl-watcher-learning-ready + if: github.event.schedule == '23 6 * * *' && steps.fetch-metrics.outputs.downloaded == 'true' + env: + GH_TOKEN: ${{ steps.source_app_token.outputs.token || github.token }} + SOURCE_GITHUB_APP_ID: ${{ vars.CROSS_REPO_GITHUB_APP_ID }} + working-directory: bridge + run: | + set -euo pipefail + output=data/output/strategy_optimization_watcher/soxl-watcher-learning-ready.json + python -m scripts.run_soxl_manual_learning \ + --watcher-result data/output/strategy_optimization_watcher/result.json \ + --diagnosis-result data/output/strategy_optimization_watcher/research-diagnosis.json \ + --github-app-id "$SOURCE_GITHUB_APP_ID" \ + --watcher-preflight | tee "$output" + python - "$output" "$GITHUB_OUTPUT" <<'PY' + import json + import sys + + value = json.load(open(sys.argv[1], encoding="utf-8")) + ready = value.get("ready") is True + with open(sys.argv[2], "a", encoding="utf-8") as target: + target.write(f"ready={'true' if ready else 'false'}\n") + if ready: + target.write(f"p1_manifest_sha256={value['p1_manifest_sha256']}\n") + PY + - name: Extract bounded research task source snapshot if: success() working-directory: bridge @@ -355,3 +387,258 @@ jobs: name: strategy-optimization-watcher-${{ github.run_id }} path: bridge/data/output/strategy_optimization_watcher/ if-no-files-found: warn + + soxl-watcher-learning: + needs: strategy-optimization-watcher + if: >- + needs.strategy-optimization-watcher.outputs.soxl_learning_ready == 'true' && + github.repository == 'QuantStrategyLab/AIAuditBridge' && + github.ref == 'refs/heads/main' && + github.event_name == 'schedule' && + vars.SOXL_WATCHER_GCP_WIF_PROVIDER != '' && + vars.SOXL_WATCHER_GCP_PROJECT_ID != '' + runs-on: [self-hosted, codex-vps] + timeout-minutes: 35 + permissions: + contents: read + actions: read + id-token: write + env: + P1_MANIFEST_SHA256: ${{ needs.strategy-optimization-watcher.outputs.soxl_p1_manifest_sha256 }} + LEARNING_ROOT: ${{ github.workspace }}/.aab-soxl-watcher-learning-${{ github.run_id }}-${{ github.run_attempt }} + steps: + - name: Check dedicated scheduled research identity configuration + id: research-identity + env: + WIF_PROVIDER: ${{ vars.SOXL_WATCHER_GCP_WIF_PROVIDER }} + GCP_PROJECT_ID: ${{ vars.SOXL_WATCHER_GCP_PROJECT_ID }} + run: | + set -euo pipefail + if [ -n "${WIF_PROVIDER:-}" ] && [ -n "${GCP_PROJECT_ID:-}" ]; then + echo 'configured=true' >> "$GITHUB_OUTPUT" + else + echo 'configured=false' >> "$GITHUB_OUTPUT" + echo 'SOXL_WATCHER_LEARNING_STATUS=PARKED_IDENTITY_NOT_CONFIGURED' >> "$GITHUB_STEP_SUMMARY" + fi + + - name: Checkout Bridge + if: steps.research-identity.outputs.configured == 'true' + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 + with: + path: bridge + persist-credentials: false + + - name: Download sanitized watcher handoff + if: steps.research-identity.outputs.configured == 'true' + uses: actions/download-artifact@v8 + with: + name: strategy-optimization-watcher-${{ github.run_id }} + path: watcher-handoff + + - name: Create source issue token + id: source-app-token + if: steps.research-identity.outputs.configured == 'true' + uses: actions/create-github-app-token@v3.2.0 + with: + app-id: ${{ vars.CROSS_REPO_GITHUB_APP_ID }} + private-key: ${{ secrets.CROSS_REPO_GITHUB_APP_PRIVATE_KEY }} + owner: QuantStrategyLab + repositories: UsEquitySnapshotPipelines + permission-issues: write + + - name: Initialize sanitized watcher learning terminal + if: steps.research-identity.outputs.configured == 'true' + run: | + set -euo pipefail + umask 077 + mkdir -p "$LEARNING_ROOT" + chmod 700 "$LEARNING_ROOT" + printf '%s\n' '{"schema_version":"qsl.soxl-manual-learning-run.v1","operation":"soxl_watcher_learning","status":"parked","failure_stage":"setup_incomplete","research_executed":false,"learning_only":true,"no_order":true,"size_zero_required":true,"promotion_eligible":false}' > "$LEARNING_ROOT/summary.json" + + - name: Checkout bounded learning consumer + if: steps.research-identity.outputs.configured == 'true' + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 + with: + repository: QuantStrategyLab/UsEquitySnapshotPipelines + ref: b03ecbe4e0a7a0de22f298499f867a7039e4b60a + path: consumer-source + persist-credentials: false + + - name: Checkout frozen strategy runtime + if: steps.research-identity.outputs.configured == 'true' + uses: actions/checkout@df4cb1c069e1874edd31b4311f1884172cec0e10 + with: + repository: QuantStrategyLab/UsEquityStrategies + ref: 7756fe32585e85cf1d09a163203a02e3eee39fe1 + path: ues-source + persist-credentials: false + + - uses: actions/setup-python@a26af69be951a213d495a4c3e4e4022e16d87065 + if: steps.research-identity.outputs.configured == 'true' + with: + python-version: '3.11' + + - name: Install frozen numeric runtimes + if: steps.research-identity.outputs.configured == 'true' + run: | + set -euo pipefail + python -m pip install --quiet 'uv==0.11.19' + cd consumer-source + env -u UV_PYTHON -u VIRTUAL_ENV uv sync --locked --no-dev --no-editable --python 3.11 + cd ../ues-source + UV_PROJECT_ENVIRONMENT="$LEARNING_ROOT/ues-env" env -u UV_PYTHON -u VIRTUAL_ENV uv sync --locked --no-dev --no-editable --python 3.11 + rm -rf -- build + test -z "$(git status --porcelain --untracked-files=all)" + + - name: Authenticate for task-bound immutable P1 read only + if: steps.research-identity.outputs.configured == 'true' + uses: google-github-actions/auth@7c6bc770dae815cd3e89ee6cdf493a5fab2cc093 + with: + project_id: ${{ vars.SOXL_WATCHER_GCP_PROJECT_ID }} + workload_identity_provider: ${{ vars.SOXL_WATCHER_GCP_WIF_PROVIDER }} + create_credentials_file: true + export_environment_variables: true + cleanup_credentials: true + + - name: Read and verify exact four-member P1 root + if: steps.research-identity.outputs.configured == 'true' + env: + CLOUDSDK_CORE_DISABLE_PROMPTS: '1' + CLOUDSDK_CORE_DISABLE_USAGE_REPORTING: 'true' + run: | + set -euo pipefail + umask 077 + root="$LEARNING_ROOT/p1-root" + completion="$LEARNING_ROOT/p1-complete.json" + source="gs://qsl-runtime-logs-shared/soxl-p1-p3/${P1_MANIFEST_SHA256}" + mkdir -p "$root" + chmod 700 "$LEARNING_ROOT" "$root" + SOURCE="$source" ROOT="$root" COMPLETION="$completion" python3 - <<'PY' + import json + import os + import stat + import subprocess + from pathlib import Path + + def fail(reason): + print(json.dumps({"status": "unavailable", "reason": reason})) + raise SystemExit(1) + + credential = os.environ.get("GOOGLE_APPLICATION_CREDENTIALS", "") + override = os.environ.get("CLOUDSDK_AUTH_CREDENTIAL_FILE_OVERRIDE", "") + try: + credential_path = Path(credential) + if not credential or override != credential or credential_path.is_symlink() or not credential_path.is_file(): + fail("temporary_identity_invalid") + if json.loads(credential_path.read_text())["type"] != "external_account": + fail("temporary_identity_invalid") + except (OSError, KeyError, TypeError, ValueError, json.JSONDecodeError): + fail("temporary_identity_invalid") + + source = os.environ["SOURCE"].rstrip("/") + "/" + members = { + "binding.json": (Path(os.environ["ROOT"]) / "binding.json", 1024 * 1024), + "bars.json": (Path(os.environ["ROOT"]) / "bars.json", 32 * 1024 * 1024), + "manifest.json": (Path(os.environ["ROOT"]) / "manifest.json", 1024 * 1024), + "p1-complete.json": (Path(os.environ["COMPLETION"]), 1024 * 1024), + } + sizes = {} + for name, (_, maximum) in members.items(): + try: + result = subprocess.run( + ["gcloud", "storage", "objects", "describe", source + name, "--raw", "--format=json"], + capture_output=True, text=True, timeout=30, check=False, + ) + except (OSError, subprocess.SubprocessError): + fail("p1_metadata_unavailable") + if result.returncode: + fail("p1_metadata_unavailable") + try: + size = int(json.loads(result.stdout)["size"]) + except (KeyError, TypeError, ValueError, json.JSONDecodeError): + fail("p1_metadata_invalid") + if size <= 0 or size > maximum: + fail("p1_member_size_invalid") + sizes[name] = size + if sum(sizes.values()) > 35 * 1024 * 1024: + fail("p1_read_budget_exceeded") + for name, (destination, _) in members.items(): + try: + result = subprocess.run( + ["gcloud", "storage", "cp", "--quiet", source + name, str(destination)], + capture_output=True, text=True, timeout=60, check=False, + ) + except (OSError, subprocess.SubprocessError): + fail("p1_object_read_unavailable") + if result.returncode: + fail("p1_object_read_unavailable") + try: + item = destination.lstat() + except OSError: + fail("p1_object_read_invalid") + if not stat.S_ISREG(item.st_mode) or item.st_size != sizes[name]: + fail("p1_object_read_invalid") + PY + ROOT="$root" COMPLETION="$completion" EXPECTED_MANIFEST_SHA256="$P1_MANIFEST_SHA256" \ + consumer-source/.venv/bin/python - <<'PY' + import os + from pathlib import Path + try: + from us_equity_snapshot_pipelines.lifecycle.soxl_core_only_p1_publisher import ( + verify_soxl_core_only_input_root, + verify_soxl_core_only_p1_remote_completion, + ) + root = Path(os.environ["ROOT"]) + expected = os.environ["EXPECTED_MANIFEST_SHA256"] + if verify_soxl_core_only_input_root(root) != expected: + raise ValueError + if verify_soxl_core_only_p1_remote_completion(root, Path(os.environ["COMPLETION"])) != expected: + raise ValueError + except Exception: + print('{"status":"unavailable","reason":"p1_validation_failed"}') + raise SystemExit(1) from None + PY + + - name: Run task-bound SOXL watcher learning consumer + if: steps.research-identity.outputs.configured == 'true' + env: + GH_TOKEN: ${{ steps.source-app-token.outputs.token }} + UV_PROJECT_ENVIRONMENT: ${{ env.LEARNING_ROOT }}/ues-env + working-directory: bridge + run: | + set -euo pipefail + python -m scripts.run_soxl_manual_learning --watcher-result \ + ../watcher-handoff/result.json \ + --diagnosis-result ../watcher-handoff/research-diagnosis.json \ + --github-app-id "${{ vars.CROSS_REPO_GITHUB_APP_ID }}" \ + --manifest-sha256 "$P1_MANIFEST_SHA256" \ + --root "$LEARNING_ROOT/p1-root" \ + --consumer-source "$GITHUB_WORKSPACE/consumer-source" \ + --ues-source "$GITHUB_WORKSPACE/ues-source" \ + --output "$LEARNING_ROOT/summary.json" + + - name: Record known pre-numeric failure for this exact task + if: failure() && steps.research-identity.outputs.configured == 'true' + env: + GH_TOKEN: ${{ steps.source-app-token.outputs.token }} + working-directory: bridge + run: | + set -euo pipefail + python -m scripts.run_soxl_manual_learning \ + --watcher-result ../watcher-handoff/result.json \ + --diagnosis-result ../watcher-handoff/research-diagnosis.json \ + --github-app-id "${{ vars.CROSS_REPO_GITHUB_APP_ID }}" \ + --watcher-record-pre-numeric-failure + + - name: Upload sanitized watcher learning terminal + if: always() && steps.research-identity.outputs.configured == 'true' + uses: actions/upload-artifact@v7 + with: + name: soxl-watcher-learning-${{ github.run_id }}-${{ github.run_attempt }} + path: ${{ env.LEARNING_ROOT }}/summary.json + if-no-files-found: error + retention-days: 35 + + - name: Remove watcher learning workspace + if: always() + run: rm -rf -- "$LEARNING_ROOT" diff --git a/scripts/run_soxl_manual_learning.py b/scripts/run_soxl_manual_learning.py index a63f8729..4c4a7f41 100644 --- a/scripts/run_soxl_manual_learning.py +++ b/scripts/run_soxl_manual_learning.py @@ -8,12 +8,24 @@ import math import re import subprocess +from urllib.parse import urlparse from collections.abc import Callable, Mapping, Sequence from pathlib import Path from typing import Any from client.config import GatewayConfig from client.gateway_client import AiGatewayClient +from service.research_diagnosis import build_research_diagnosis_request, marker_for_research_diagnosis +from service.research_task import ( + SOXL_WATCHER_CANDIDATE_ID, + SOXL_WATCHER_CONSUMER_REVISION, + SOXL_WATCHER_P2_CONFIG_SHA256, + SOXL_WATCHER_PARAMETER_BOUNDS_SHA256, + SOXL_WATCHER_QPK_REVISION, + SOXL_WATCHER_STRATEGY_REPOSITORY, + SOXL_WATCHER_UES_REVISION, + validate_strategy_diagnosis_task, +) ADVICE_SCHEMA = "qsl.soxl-manual-learning-advice.v1" ARTIFACT_SCHEMA = "qsl.soxl-manual-learning-run.v1" @@ -28,6 +40,7 @@ EXPECTED_REF = "refs/heads/main" EXPECTED_EVENT = "workflow_dispatch" SAFE_REASON = re.compile(r"[a-z0-9_]{1,64}\Z") +WATCHER_MARKER_PREFIX = "qsl-soxl-watcher-learning:v1" class ManualLearningError(ValueError): @@ -212,8 +225,7 @@ def _sanitize_numeric(value: object, values: Sequence[float], manifest_sha256: s not isinstance(source, Mapping) or source.get("repository") != "QuantStrategyLab/UsEquityStrategies" or source.get("revision") != UES_REVISION - or not isinstance(source.get("quant_platform_kit_revision"), str) - or re.fullmatch(r"[0-9a-f]{40}", source["quant_platform_kit_revision"]) is None + or source.get("quant_platform_kit_revision") != SOXL_WATCHER_QPK_REVISION or not isinstance(source.get("uv_lock_sha256"), str) or re.fullmatch(r"[0-9a-f]{64}", source["uv_lock_sha256"]) is None ): @@ -270,6 +282,370 @@ def _sanitize_numeric(value: object, values: Sequence[float], manifest_sha256: s return safe, safe_source, digest +def _issue_number(repository: str, issue_url: str) -> str: + parsed = urlparse(issue_url) + parts = parsed.path.strip("/").split("/") + if ( + parsed.scheme != "https" or parsed.netloc != "github.com" + or len(parts) != 4 or "/".join(parts[:2]) != repository + or parts[2] != "issues" or not parts[3].isdigit() + ): + raise ManualLearningError("watcher_issue_invalid") + return parts[3] + + +def read_issue_comments(repository: str, issue_url: str) -> list[dict[str, Any]]: + completed = subprocess.run( + ["gh", "api", "--paginate", "--slurp", f"repos/{repository}/issues/{_issue_number(repository, issue_url)}/comments?per_page=100"], + capture_output=True, text=True, timeout=30, check=False, + ) + if completed.returncode: + raise ManualLearningError("watcher_comments_unavailable") + try: + value = json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise ManualLearningError("watcher_comments_unavailable") from exc + if not isinstance(value, list) or any(not isinstance(page, list) for page in value): + raise ManualLearningError("watcher_comments_unavailable") + comments = [item for page in value for item in page] + if any(not isinstance(item, dict) for item in comments): + raise ManualLearningError("watcher_comments_unavailable") + return comments + + +def write_issue_comment(repository: str, issue_url: str, body: str) -> str: + completed = subprocess.run( + ["gh", "issue", "comment", issue_url, "--repo", repository, "--body-file", "-"], + input=body, capture_output=True, text=True, timeout=30, check=False, + ) + if completed.returncode: + raise ManualLearningError("watcher_comment_write_failed") + return completed.stdout.strip() + + +def _trusted_comment_bodies(comments: object, github_app_id: str) -> list[str]: + if not github_app_id.isdigit() or int(github_app_id) <= 0: + raise ManualLearningError("watcher_comment_identity_invalid") + if not isinstance(comments, list): + raise ManualLearningError("watcher_comments_unavailable") + bodies: list[str] = [] + for comment in comments: + if not isinstance(comment, Mapping): + raise ManualLearningError("watcher_comments_unavailable") + app = comment.get("performed_via_github_app") + if isinstance(app, Mapping) and app.get("id") == int(github_app_id) and isinstance(comment.get("body"), str): + bodies.append(comment["body"]) + return bodies + + +def watcher_learning_comment( + task: Mapping[str, Any], *, phase: str, status: str, + numeric_result_sha256: str = "", numeric_summary: object = None, + numeric_source_identity: object = None, failure_stage: str | None = None, +) -> str: + verified = validate_strategy_diagnosis_task(task) + if phase not in {"started", "terminal"} or status not in {"started", "accepted", "failed"}: + raise ManualLearningError("watcher_stage_invalid") + if phase == "started" and status != "started": + raise ManualLearningError("watcher_stage_invalid") + if phase == "terminal" and status == "started": + raise ManualLearningError("watcher_stage_invalid") + if failure_stage not in {None, "pre_numeric_failed", "numeric_execution_failed"}: + raise ManualLearningError("watcher_stage_invalid") + if status == "failed" and failure_stage is None: + failure_stage = "numeric_execution_failed" + if status != "failed" and failure_stage is not None: + raise ManualLearningError("watcher_stage_invalid") + if numeric_result_sha256 and re.fullmatch(r"[0-9a-f]{64}", numeric_result_sha256) is None: + raise ManualLearningError("numeric_result_invalid") + marker = ":".join( + ( + WATCHER_MARKER_PREFIX, phase, verified["task_sha256"], + SOXL_WATCHER_PARAMETER_BOUNDS_SHA256, SOXL_WATCHER_CONSUMER_REVISION, + ) + ) + record = { + "task_id": verified["task_id"], "task_sha256": verified["task_sha256"], + "parameter_bounds_sha256": SOXL_WATCHER_PARAMETER_BOUNDS_SHA256, + "consumer_revision": SOXL_WATCHER_CONSUMER_REVISION, + "strategy_revision": SOXL_WATCHER_UES_REVISION, + "status": status, "numeric_result_sha256": numeric_result_sha256 or None, + "numeric_summary": numeric_summary, + "numeric_source_identity": numeric_source_identity, + "failure_stage": failure_stage, + "learning_only": True, "no_order": True, "promotion_eligible": False, + } + return f"\n`{canonical_json(record)}`" + + +def canonical_json(value: object) -> str: + return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False) + + +def _marker(task: Mapping[str, Any], phase: str) -> str: + return ":".join( + (WATCHER_MARKER_PREFIX, phase, str(task["task_sha256"]), + SOXL_WATCHER_PARAMETER_BOUNDS_SHA256, SOXL_WATCHER_CONSUMER_REVISION) + ) + + +def _terminal_from_comments(task: Mapping[str, Any], bodies: Sequence[str]) -> dict[str, Any] | None: + marker = f"" + candidates = [body for body in bodies if body.startswith(marker)] + if not candidates: + return None + if any(not body.startswith(marker + "\n`") or not body.endswith("`") for body in candidates): + raise ManualLearningError("watcher_terminal_invalid") + parsed: list[dict[str, Any]] = [] + for body in candidates: + try: + value = json.loads(body[len(marker) + 2 : -1]) + except json.JSONDecodeError as exc: + raise ManualLearningError("watcher_terminal_invalid") from exc + expected = { + "task_id", "task_sha256", "parameter_bounds_sha256", "consumer_revision", + "strategy_revision", "status", "numeric_result_sha256", "learning_only", + "numeric_summary", "numeric_source_identity", "failure_stage", "no_order", "promotion_eligible", + } + if not isinstance(value, dict) or set(value) != expected: + raise ManualLearningError("watcher_terminal_invalid") + if ( + value["task_id"] != task["task_id"] or value["task_sha256"] != task["task_sha256"] + or value["parameter_bounds_sha256"] != SOXL_WATCHER_PARAMETER_BOUNDS_SHA256 + or value["consumer_revision"] != SOXL_WATCHER_CONSUMER_REVISION + or value["strategy_revision"] != SOXL_WATCHER_UES_REVISION + or value["status"] not in {"accepted", "failed"} + or value["failure_stage"] not in {None, "pre_numeric_failed", "numeric_execution_failed"} + or (value["status"] == "accepted" and value["failure_stage"] is not None) + or (value["status"] == "failed" and value["failure_stage"] is None) + or value["learning_only"] is not True or value["no_order"] is not True + or value["promotion_eligible"] is not False + or (value["status"] == "accepted" and re.fullmatch(r"[0-9a-f]{64}", str(value["numeric_result_sha256"] or "")) is None) + ): + raise ManualLearningError("watcher_terminal_invalid") + if value["status"] == "accepted": + raw_results = [] + if not isinstance(value["numeric_summary"], list): + raise ManualLearningError("watcher_terminal_invalid") + for item in value["numeric_summary"]: + if not isinstance(item, Mapping): + raise ManualLearningError("watcher_terminal_invalid") + raw_results.append({ + "schema_version": REPLAY_SCHEMA, "status": "SUCCESS", + "parameter_override": item.get("parameter_override"), + "cost_bps": item.get("cost_bps"), + "backtest_result": item.get("backtest_result"), + "output_sha256": item.get("output_sha256"), + }) + reconstructed = { + "schema_version": NUMERIC_SCHEMA, "status": "SUCCESS", + "learning_only": True, "no_order": True, "size_zero_required": True, + "promotion_eligible": False, "research_executed": True, + "development_cutoff": DEVELOPMENT_CUTOFF, + "p1_identity": {"input_manifest_sha256": task["evidence"]["p1_input_digest"]}, + "source_identity": value["numeric_source_identity"], + "parameter_key": "blend_gate_mid_soxl_weight", "trial_count": 3, + "cost_bps": list(COST_BPS), "results": raw_results, + "result_sha256": value["numeric_result_sha256"], + } + safe, source, digest = _sanitize_numeric( + reconstructed, (0.65, 0.6, 0.55), task["evidence"]["p1_input_digest"], + ) + value["numeric_summary"] = safe + value["numeric_source_identity"] = source + value["numeric_result_sha256"] = digest + elif value["numeric_summary"] is not None or value["numeric_source_identity"] is not None or value["numeric_result_sha256"] is not None: + raise ManualLearningError("watcher_terminal_invalid") + parsed.append(value) + if any(value != parsed[0] for value in parsed[1:]): + raise ManualLearningError("watcher_terminal_conflict") + return parsed[0] + + +def _watcher_context( + watcher_result: Mapping[str, Any], diagnosis_result: Mapping[str, Any], + *, github_app_id: str, + read_comments: Callable[[str, str], list[dict[str, Any]]], +) -> dict[str, Any]: + snapshot = watcher_result.get("research_task_source_snapshot") + issues = watcher_result.get("issues") + if not isinstance(snapshot, Mapping) or snapshot.get("data_status") != "ready" or not isinstance(issues, list): + raise ManualLearningError("watcher_task_unavailable") + tasks = snapshot.get("tasks") + if not isinstance(tasks, list) or len(tasks) != 1 or not isinstance(tasks[0], Mapping): + raise ManualLearningError("watcher_task_unavailable") + task = validate_strategy_diagnosis_task(tasks[0]) + if ( + task["target"] != { + "candidate_id": SOXL_WATCHER_CANDIDATE_ID, "candidate_kind": "individual", + "domain": "us_equity", "repository": SOXL_WATCHER_STRATEGY_REPOSITORY, + "strategy_revision": SOXL_WATCHER_UES_REVISION, + } + or task["evidence"]["p2_config_digest"] != SOXL_WATCHER_P2_CONFIG_SHA256 + or task["experiment"]["parameter_bounds_sha256"] != SOXL_WATCHER_PARAMETER_BOUNDS_SHA256 + ): + raise ManualLearningError("watcher_task_not_executable") + event_key = task["task_id"].removeprefix("watcher-") + matching = [ + issue for issue in issues if isinstance(issue, Mapping) + and isinstance(issue.get("task"), Mapping) and issue["task"].get("event_key") == event_key + and issue.get("repo") == "QuantStrategyLab/UsEquitySnapshotPipelines" + and isinstance(issue.get("url") or issue.get("existing_url"), str) + ] + if len(matching) != 1: + raise ManualLearningError("watcher_issue_unavailable") + repository = str(matching[0]["repo"]) + issue_url = str(matching[0].get("url") or matching[0].get("existing_url")) + _issue_number(repository, issue_url) + bodies = _trusted_comment_bodies(read_comments(repository, issue_url), github_app_id) + if not isinstance(diagnosis_result.get("diagnoses"), list): + raise ManualLearningError("watcher_diagnosis_unavailable") + diagnosis_marker = marker_for_research_diagnosis(build_research_diagnosis_request(task)) + if not any(body.startswith(diagnosis_marker) for body in bodies): + raise ManualLearningError("watcher_diagnosis_unavailable") + terminal = _terminal_from_comments(task, bodies) + started = any(body.startswith(f"\n") for body in bodies) + return {"task": task, "repository": repository, "issue_url": issue_url, "terminal": terminal, "started": started} + + +def prepare_watcher_learning( + watcher_result: Mapping[str, Any], diagnosis_result: Mapping[str, Any], *, github_app_id: str, + read_comments: Callable[[str, str], list[dict[str, Any]]] = read_issue_comments, +) -> dict[str, Any]: + try: + context = _watcher_context( + watcher_result, diagnosis_result, github_app_id=github_app_id, read_comments=read_comments, + ) + except (ManualLearningError, OSError, ValueError, subprocess.SubprocessError): + return {"status": "parked", "ready": False} + if context["terminal"] is not None: + return {"status": "reused", "ready": False} + if context["started"]: + return {"status": "parked", "ready": False, "failure_stage": "numeric_outcome_unknown"} + task = context["task"] + return { + "status": "ready", "ready": True, "p1_manifest_sha256": task["evidence"]["p1_input_digest"], + "producer_revision": task["evidence"]["producer_revision"], + } + + +def _watcher_base(task: Mapping[str, Any], manifest_sha256: str) -> dict[str, Any]: + return { + "schema_version": ARTIFACT_SCHEMA, "operation": "soxl_watcher_learning", + "source": "watcher_event_independent_learning", "status": "parked", + "task_id": task["task_id"], "task_sha256": task["task_sha256"], + "experiment": {"parameter_bounds_sha256": SOXL_WATCHER_PARAMETER_BOUNDS_SHA256}, + "parameter_key": "blend_gate_mid_soxl_weight", "parameter_values": [0.65, 0.6, 0.55], + "cost_bps": list(COST_BPS), "development_cutoff": DEVELOPMENT_CUTOFF, + "input_identity": {"manifest_sha256": manifest_sha256, "member_count": 4}, + "consumer_source": {"repository": "QuantStrategyLab/UsEquitySnapshotPipelines", "revision": SOXL_WATCHER_CONSUMER_REVISION}, + "learning_only": True, "no_order": True, "size_zero_required": True, + "promotion_eligible": False, "research_executed": False, + } + + +def record_watcher_pre_numeric_failure( + *, watcher_result: Mapping[str, Any], diagnosis_result: Mapping[str, Any], github_app_id: str, + read_comments: Callable[[str, str], list[dict[str, Any]]] = read_issue_comments, + write_comment: Callable[[str, str, str], str] = write_issue_comment, +) -> dict[str, Any]: + context = _watcher_context( + watcher_result, diagnosis_result, github_app_id=github_app_id, read_comments=read_comments, + ) + if context["terminal"] is not None: + return {"status": "reused"} + if context["started"]: + raise ManualLearningError("numeric_outcome_unknown") + write_comment( + context["repository"], context["issue_url"], + watcher_learning_comment( + context["task"], phase="terminal", status="failed", failure_stage="pre_numeric_failed", + ), + ) + return {"status": "failed", "failure_stage": "pre_numeric_failed"} + + +def run_watcher_learning( + *, watcher_result: Mapping[str, Any], diagnosis_result: Mapping[str, Any], manifest_sha256: str, + root: Path, consumer_source: Path, ues_source: Path, github_app_id: str, + read_comments: Callable[[str, str], list[dict[str, Any]]] = read_issue_comments, + write_comment: Callable[[str, str, str], str] = write_issue_comment, + command_runner: Callable[[list[str]], Any] | None = None, +) -> dict[str, Any]: + try: + context = _watcher_context(watcher_result, diagnosis_result, github_app_id=github_app_id, read_comments=read_comments) + except ManualLearningError as exc: + return {"status": "parked", "failure_stage": str(exc), "research_executed": False, + "learning_only": True, "no_order": True, "size_zero_required": True, "promotion_eligible": False} + except (OSError, ValueError, subprocess.SubprocessError): + return {"status": "parked", "failure_stage": "watcher_comments_unavailable", "research_executed": False, + "learning_only": True, "no_order": True, "size_zero_required": True, "promotion_eligible": False} + task = context["task"] + artifact = _watcher_base(task, manifest_sha256) + if manifest_sha256 != task["evidence"]["p1_input_digest"]: + artifact["failure_stage"] = "input_identity_invalid" + return artifact + terminal = context["terminal"] + if terminal is not None: + artifact["status"] = "accepted" if terminal["status"] == "accepted" else "parked" + artifact["research_executed"] = terminal["status"] == "accepted" + artifact["numeric_execution"] = {"status": "reused"} + artifact["numeric_result_sha256"] = terminal["numeric_result_sha256"] + if terminal["status"] == "failed": + artifact["failure_stage"] = terminal["failure_stage"] + if terminal["status"] == "accepted": + artifact["numeric_summary"] = terminal["numeric_summary"] + artifact["numeric_source_identity"] = terminal["numeric_source_identity"] + return artifact + if context["started"]: + artifact.update(failure_stage="numeric_outcome_unknown", research_executed=None, numeric_execution={"status": "outcome_unknown"}) + return artifact + required = ( + root / "binding.json", root / "manifest.json", root / "bars.json", + consumer_source / "scripts/run_soxl_three_asset_learning.py", + consumer_source / "config/soxl_soxx_core_only_p2_v3.json", + ) + if any(path.is_symlink() or not path.is_file() for path in required) or not (consumer_source / ".venv/bin/python").is_file() or not ues_source.is_dir(): + artifact["failure_stage"] = "source_or_input_unavailable" + return artifact + try: + write_comment(context["repository"], context["issue_url"], watcher_learning_comment(task, phase="started", status="started")) + except (ManualLearningError, OSError, subprocess.SubprocessError): + artifact["failure_stage"] = "watcher_comment_write_failed" + return artifact + artifact.update(research_executed=None, numeric_execution={"status": "started"}) + runner = command_runner or (lambda argv: subprocess.run(argv, capture_output=True, text=True, timeout=1200, check=False)) + try: + completed = runner(_numeric_command(root, consumer_source, ues_source, (0.65, 0.6, 0.55))) + except (OSError, subprocess.SubprocessError): + artifact.update(failure_stage="numeric_outcome_unknown", numeric_execution={"status": "outcome_unknown"}) + return artifact + if getattr(completed, "returncode", None) != 0: + artifact.update(failure_stage="numeric_execution_failed", numeric_execution={"status": "failed"}) + try: + write_comment(context["repository"], context["issue_url"], watcher_learning_comment(task, phase="terminal", status="failed")) + except (ManualLearningError, OSError, subprocess.SubprocessError): + artifact["failure_stage"] = "numeric_outcome_unknown" + artifact["numeric_execution"] = {"status": "outcome_unknown"} + return artifact + try: + safe, source, digest = _sanitize_numeric(json.loads(completed.stdout), (0.65, 0.6, 0.55), manifest_sha256) + write_comment( + context["repository"], context["issue_url"], + watcher_learning_comment( + task, phase="terminal", status="accepted", numeric_result_sha256=digest, + numeric_summary=safe, numeric_source_identity=source, + ), + ) + except (AttributeError, TypeError, json.JSONDecodeError, ManualLearningError, OSError, subprocess.SubprocessError): + artifact.update(failure_stage="numeric_outcome_unknown", numeric_execution={"status": "outcome_unknown"}) + return artifact + artifact.update(status="accepted", research_executed=True, numeric_summary=safe, + numeric_source_identity=source, numeric_result_sha256=digest, + numeric_execution={"status": "succeeded"}) + return artifact + + def run_manual_learning( *, parameter_grid: str, manifest_sha256: str, root: Path, consumer_source: Path, ues_source: Path, context: Mapping[str, str], @@ -382,19 +758,64 @@ def initialize_record(path: Path, context: Mapping[str, str]) -> None: def main(argv: Sequence[str] | None = None) -> int: parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--parameter-grid", required=True) - parser.add_argument("--manifest-sha256", required=True) - parser.add_argument("--root", required=True, type=Path) - parser.add_argument("--consumer-source", required=True, type=Path) - parser.add_argument("--ues-source", required=True, type=Path) - parser.add_argument("--output", required=True, type=Path) - parser.add_argument("--repository", required=True) - parser.add_argument("--ref", required=True) - parser.add_argument("--event-name", required=True) - parser.add_argument("--actor", required=True) - parser.add_argument("--run-id", required=True) - parser.add_argument("--run-attempt", required=True) + parser.add_argument("--parameter-grid") + parser.add_argument("--manifest-sha256") + parser.add_argument("--root", type=Path) + parser.add_argument("--consumer-source", type=Path) + parser.add_argument("--ues-source", type=Path) + parser.add_argument("--output", type=Path) + parser.add_argument("--repository") + parser.add_argument("--ref") + parser.add_argument("--event-name") + parser.add_argument("--actor") + parser.add_argument("--run-id") + parser.add_argument("--run-attempt") + parser.add_argument("--watcher-result", type=Path) + parser.add_argument("--diagnosis-result", type=Path) + parser.add_argument("--github-app-id") + parser.add_argument("--watcher-preflight", action="store_true") + parser.add_argument("--watcher-record-pre-numeric-failure", action="store_true") args = parser.parse_args(argv) + if args.watcher_result is not None: + if args.diagnosis_result is None or not args.github_app_id: + parser.error("watcher mode requires --diagnosis-result and --github-app-id") + watcher_result = json.loads(args.watcher_result.read_text(encoding="utf-8")) + diagnosis_result = json.loads(args.diagnosis_result.read_text(encoding="utf-8")) + if args.watcher_preflight: + result = prepare_watcher_learning( + watcher_result, diagnosis_result, github_app_id=args.github_app_id, + ) + print(canonical_json(result)) + return 0 + if args.watcher_record_pre_numeric_failure: + try: + result = record_watcher_pre_numeric_failure( + watcher_result=watcher_result, diagnosis_result=diagnosis_result, + github_app_id=args.github_app_id, + ) + except (ManualLearningError, OSError, ValueError, subprocess.SubprocessError): + print(canonical_json({"status": "parked", "failure_stage": "watcher_failure_record_unavailable"})) + return 2 + print(canonical_json(result)) + return 0 if result["status"] in {"failed", "reused"} else 2 + if any(value is None for value in (args.manifest_sha256, args.root, args.consumer_source, args.ues_source, args.output)): + parser.error("watcher execution requires manifest, source, root and output paths") + result = run_watcher_learning( + watcher_result=watcher_result, diagnosis_result=diagnosis_result, + manifest_sha256=args.manifest_sha256, root=args.root, + consumer_source=args.consumer_source, ues_source=args.ues_source, + github_app_id=args.github_app_id, + ) + _write(args.output, result) + print(json.dumps({"status": result["status"], "operation": "soxl_watcher_learning"}, sort_keys=True)) + return 0 if result["status"] == "accepted" else 2 + manual_required = ( + args.parameter_grid, args.manifest_sha256, args.root, args.consumer_source, + args.ues_source, args.output, args.repository, args.ref, args.event_name, + args.actor, args.run_id, args.run_attempt, + ) + if any(value is None for value in manual_required): + parser.error("manual mode requires the original bounded input and authority arguments") context = {key: getattr(args, key) for key in ("repository", "ref", "event_name", "actor", "run_id", "run_attempt")} try: result = run_manual_learning( diff --git a/service/research_task.py b/service/research_task.py index 05e4173e..321ca782 100644 --- a/service/research_task.py +++ b/service/research_task.py @@ -25,6 +25,21 @@ _CANDIDATE_KINDS = frozenset({"individual", "portfolio", "plugin"}) _DOMAINS = frozenset({"us_equity", "hk_equity", "cn_equity", "crypto"}) _EVIDENCE_FIELDS = frozenset({"p1_input_digest", "p2_config_digest", "p3_evidence_id", "strategy_revision", "producer_revision"}) +SOXL_WATCHER_CANDIDATE_ID = "soxl_soxx_core_only_p2_v3" +SOXL_WATCHER_STRATEGY_REPOSITORY = "QuantStrategyLab/UsEquityStrategies" +SOXL_WATCHER_UES_REVISION = "7756fe32585e85cf1d09a163203a02e3eee39fe1" +SOXL_WATCHER_P2_CONFIG_SHA256 = "ff8fa0acf4f175a7c40c3e1e6a3304ea2748b6b81c3797342085a4df3810ab4d" +SOXL_WATCHER_CONSUMER_REVISION = "b03ecbe4e0a7a0de22f298499f867a7039e4b60a" +SOXL_WATCHER_QPK_REVISION = "3acab1923a97b805b077c85c6c19657be0143bac" +SOXL_WATCHER_PARAMETER_BOUNDS = { + "parameter_key": "blend_gate_mid_soxl_weight", + "parameter_values": [0.65, 0.60, 0.55], + "cost_bps": [5.0, 10.0, 15.0], + "development_cutoff": "2025-07-31", + "consumer_revision": SOXL_WATCHER_CONSUMER_REVISION, + "strategy_revision": SOXL_WATCHER_UES_REVISION, + "quant_platform_kit_revision": SOXL_WATCHER_QPK_REVISION, +} class ResearchTaskError(ValueError): @@ -38,6 +53,25 @@ def canonical_json(value: object) -> str: raise ResearchTaskError("research task must use finite JSON values") from exc +SOXL_WATCHER_PARAMETER_BOUNDS_SHA256 = hashlib.sha256( + canonical_json(SOXL_WATCHER_PARAMETER_BOUNDS).encode("utf-8") +).hexdigest() + + +def _is_exact_soxl_watcher_target( + *, candidate_id: object, candidate_kind: object, domain: object, + strategy_repository: object, evidence: Mapping[str, str], +) -> bool: + return ( + candidate_id == SOXL_WATCHER_CANDIDATE_ID + and candidate_kind == "individual" + and domain == "us_equity" + and strategy_repository == SOXL_WATCHER_STRATEGY_REPOSITORY + and evidence.get("strategy_revision") == SOXL_WATCHER_UES_REVISION + and evidence.get("p2_config_digest") == SOXL_WATCHER_P2_CONFIG_SHA256 + ) + + def calculate_task_sha256(payload: Mapping[str, Any]) -> str: material = copy.deepcopy(dict(payload)) material.pop("task_sha256", None) @@ -108,6 +142,14 @@ def build_strategy_diagnosis_task( if not isinstance(strategy_repository, str) or not _REPOSITORY.fullmatch(strategy_repository): raise ResearchTaskError("strategy repository is invalid") verified_evidence = _evidence(evidence) + parameter_bounds_sha256 = ( + SOXL_WATCHER_PARAMETER_BOUNDS_SHA256 + if _is_exact_soxl_watcher_target( + candidate_id=candidate, candidate_kind=candidate_kind, domain=domain, + strategy_repository=strategy_repository, evidence=verified_evidence, + ) + else None + ) task: dict[str, Any] = { "schema": SCHEMA, "task_id": f"watcher-{event_key}", @@ -130,7 +172,7 @@ def build_strategy_diagnosis_task( "experiment": { "objective": "diagnose_degradation", "hypothesis": "A verified P3 observation crossed a degradation threshold; diagnose it with one bounded offline comparison without changing active parameters.", - "parameter_bounds_sha256": None, + "parameter_bounds_sha256": parameter_bounds_sha256, "max_runs": 1, "max_wall_seconds": 3600, }, @@ -216,11 +258,21 @@ def validate_strategy_diagnosis_task(value: Mapping[str, Any]) -> dict[str, Any] } if not isinstance(experiment, Mapping) or set(experiment) != expected_experiment: raise ResearchTaskError("research task experiment is incomplete") + parameter_bounds_sha256 = experiment.get("parameter_bounds_sha256") + exact_soxl_target = _is_exact_soxl_watcher_target( + candidate_id=candidate_id, candidate_kind=candidate_kind, domain=domain, + strategy_repository=repository, evidence=evidence, + ) + if parameter_bounds_sha256 is not None: + _sha256(parameter_bounds_sha256, "experiment.parameter_bounds_sha256") if ( experiment.get("objective") != "diagnose_degradation" or experiment.get("hypothesis") != "A verified P3 observation crossed a degradation threshold; diagnose it with one bounded offline comparison without changing active parameters." - or experiment.get("parameter_bounds_sha256") is not None + or parameter_bounds_sha256 not in ( + {None, SOXL_WATCHER_PARAMETER_BOUNDS_SHA256} + if exact_soxl_target else {None} + ) or experiment.get("max_runs") != 1 or experiment.get("max_wall_seconds") != 3600 ): @@ -267,6 +319,14 @@ def validate_strategy_diagnosis_task(value: Mapping[str, Any]) -> dict[str, Any] __all__ = [ "ResearchTaskError", "SCHEMA", + "SOXL_WATCHER_CANDIDATE_ID", + "SOXL_WATCHER_CONSUMER_REVISION", + "SOXL_WATCHER_P2_CONFIG_SHA256", + "SOXL_WATCHER_QPK_REVISION", + "SOXL_WATCHER_PARAMETER_BOUNDS", + "SOXL_WATCHER_PARAMETER_BOUNDS_SHA256", + "SOXL_WATCHER_STRATEGY_REPOSITORY", + "SOXL_WATCHER_UES_REVISION", "build_strategy_diagnosis_task", "calculate_task_sha256", "validate_strategy_diagnosis_task", diff --git a/tests/test_run_soxl_manual_learning.py b/tests/test_run_soxl_manual_learning.py index 411d6e51..b831b024 100644 --- a/tests/test_run_soxl_manual_learning.py +++ b/tests/test_run_soxl_manual_learning.py @@ -16,7 +16,12 @@ initialize_record, parse_parameter_grid, run_manual_learning, + run_watcher_learning, + read_issue_comments, + record_watcher_pre_numeric_failure, + write_issue_comment, ) +from service.research_task import SOXL_WATCHER_PARAMETER_BOUNDS_SHA256, build_strategy_diagnosis_task UES_REVISION = "7756fe32585e85cf1d09a163203a02e3eee39fe1" @@ -144,6 +149,223 @@ def numeric_result() -> dict[str, object]: } +def watcher_inputs() -> tuple[dict[str, object], dict[str, object]]: + task = build_strategy_diagnosis_task( + event_key="123456789abc", created_at="2026-09-10T00:00:00Z", + candidate_id="soxl_soxx_core_only_p2_v3", candidate_kind="individual", + domain="us_equity", strategy_repository="QuantStrategyLab/UsEquityStrategies", + evidence={"p1_input_digest": "0" * 64, "p2_config_digest": "ff8fa0acf4f175a7c40c3e1e6a3304ea2748b6b81c3797342085a4df3810ab4d", "p3_evidence_id": "c" * 64, "strategy_revision": UES_REVISION, "producer_revision": "e" * 40}, + ) + result = { + "research_task_source_snapshot": {"data_status": "ready", "tasks": [task]}, + "issues": [{"repo": "QuantStrategyLab/UsEquitySnapshotPipelines", "url": "https://github.com/QuantStrategyLab/UsEquitySnapshotPipelines/issues/1", "task": {"event_key": "123456789abc", "trigger": {}}}], + } + diagnosis = {"diagnoses": [{"status": "diagnosed", "task_id": task["task_id"], "task_sha256": task["task_sha256"], "issue_url": "https://github.com/QuantStrategyLab/UsEquitySnapshotPipelines/issues/1"}]} + return result, diagnosis + + +def trusted_comment(body: str) -> dict[str, object]: + return {"body": body, "performed_via_github_app": {"id": 42}} + + +def trusted_diagnosis_comment(watcher: dict[str, object]) -> dict[str, object]: + task = watcher["research_task_source_snapshot"]["tasks"][0] + marker = __import__("service.research_diagnosis", fromlist=["marker_for_research_diagnosis"]).marker_for_research_diagnosis(__import__("service.research_diagnosis", fromlist=["build_research_diagnosis_request"]).build_research_diagnosis_request(task)) + return trusted_comment(marker + "\ntrusted diagnosis") + + +def test_watcher_learning_runs_synthetic_subprocess_stub_once_and_writes_bound_terminal( + paths: dict[str, Path], +) -> None: + watcher, diagnosis = watcher_inputs() + script = paths["consumer"] / "scripts/run_soxl_three_asset_learning.py" + script.write_text("import json\nprint(json.dumps(" + repr(numeric_result()) + "))\n") + comments: list[str] = [] + artifact = run_watcher_learning( + watcher_result=watcher, diagnosis_result=diagnosis, manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], + github_app_id="42", read_comments=lambda *_: [trusted_diagnosis_comment(watcher)], + write_comment=lambda _repo, _url, body: comments.append(body) or "comment", + ) + assert artifact["status"] == "accepted" + assert artifact["source"] == "watcher_event_independent_learning" + assert artifact["promotion_eligible"] is False + assert artifact["experiment"]["parameter_bounds_sha256"] == SOXL_WATCHER_PARAMETER_BOUNDS_SHA256 + assert len(artifact["numeric_summary"]) == 9 + assert len(comments) == 2 + assert ":started:" in comments[0] and ":terminal:" in comments[1] + + +@pytest.mark.parametrize("state", ["old_null", "not_diagnosed", "started", "failed"]) +def test_watcher_learning_non_executable_or_terminal_state_never_calls_numeric(state: str, paths: dict[str, Path]) -> None: + watcher, diagnosis = watcher_inputs() + task = watcher["research_task_source_snapshot"]["tasks"][0] + comments: list[dict[str, object]] = [] + if state == "old_null": + from service.research_task import calculate_task_sha256 + task["experiment"]["parameter_bounds_sha256"] = None + task["task_sha256"] = calculate_task_sha256(task) + elif state == "not_diagnosed": + diagnosis = {"diagnoses": [{"status": "deferred", "task_id": task["task_id"]}]} + else: + diagnosis = {"status": "skipped", "diagnoses": []} + phase = "started" if state == "started" else "terminal" + body = __import__("scripts.run_soxl_manual_learning", fromlist=["watcher_learning_comment"]).watcher_learning_comment(task, phase=phase, status="failed" if state == "failed" else "started") + comments = [trusted_comment(body), trusted_comment(__import__("service.research_diagnosis", fromlist=["marker_for_research_diagnosis"]).marker_for_research_diagnosis(__import__("service.research_diagnosis", fromlist=["build_research_diagnosis_request"]).build_research_diagnosis_request(task)))] + calls: list[list[str]] = [] + artifact = run_watcher_learning( + watcher_result=watcher, diagnosis_result=diagnosis, manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], github_app_id="42", + read_comments=lambda *_: comments, write_comment=lambda *_: "comment", + command_runner=lambda argv: calls.append(argv), + ) + assert artifact["status"] in {"parked", "rejected"} + assert calls == [] + + +def test_watcher_learning_reuses_trusted_terminal_when_current_diagnosis_is_skipped(paths: dict[str, Path]) -> None: + watcher, _ = watcher_inputs() + task = watcher["research_task_source_snapshot"]["tasks"][0] + diagnosis_marker = __import__("service.research_diagnosis", fromlist=["marker_for_research_diagnosis"]).marker_for_research_diagnosis(__import__("service.research_diagnosis", fromlist=["build_research_diagnosis_request"]).build_research_diagnosis_request(task)) + raw = numeric_result() + safe = [ + {key: item[key] for key in ("parameter_override", "cost_bps", "backtest_result", "output_sha256")} + for item in raw["results"] + ] + terminal = __import__("scripts.run_soxl_manual_learning", fromlist=["watcher_learning_comment"]).watcher_learning_comment( + task, phase="terminal", status="accepted", numeric_result_sha256="b" * 64, + numeric_summary=safe, numeric_source_identity={key: raw["source_identity"][key] for key in ("repository", "revision", "quant_platform_kit_revision", "uv_lock_sha256")}, + ) + artifact = run_watcher_learning( + watcher_result=watcher, diagnosis_result={"status": "skipped", "diagnoses": []}, manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], github_app_id="42", + read_comments=lambda *_: [trusted_comment(diagnosis_marker), trusted_comment(terminal)], + write_comment=lambda *_: (_ for _ in ()).throw(AssertionError("must reuse")), + command_runner=lambda _argv: (_ for _ in ()).throw(AssertionError("must not run")), + ) + assert artifact["status"] == "accepted" + assert artifact["numeric_execution"] == {"status": "reused"} + assert len(artifact["numeric_summary"]) == 9 + + +def test_watcher_learning_rejects_unreadable_or_untrusted_comment_history(paths: dict[str, Path]) -> None: + watcher, diagnosis = watcher_inputs() + task = watcher["research_task_source_snapshot"]["tasks"][0] + diagnosis_marker = __import__("service.research_diagnosis", fromlist=["marker_for_research_diagnosis"]).marker_for_research_diagnosis(__import__("service.research_diagnosis", fromlist=["build_research_diagnosis_request"]).build_research_diagnosis_request(task)) + for active_diagnosis, reader in ( + (diagnosis, lambda *_: (_ for _ in ()).throw(OSError("private failure"))), + ({"status": "skipped", "diagnoses": []}, lambda *_: [{"body": diagnosis_marker, "performed_via_github_app": {"id": 7}}]), + ): + calls: list[list[str]] = [] + artifact = run_watcher_learning( + watcher_result=watcher, diagnosis_result=active_diagnosis, manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], + github_app_id="42", read_comments=reader, write_comment=lambda *_: "comment", + command_runner=lambda argv: calls.append(argv), + ) + assert artifact["status"] == "parked" + assert calls == [] + assert "private failure" not in json.dumps(artifact) + + +def test_comment_reader_slurps_and_flattens_every_page(monkeypatch: pytest.MonkeyPatch) -> None: + pages = [[{"body": f"comment-{index}"} for index in range(100)], [{"body": "comment-100"}]] + observed: list[list[str]] = [] + + def run(argv: list[str], **_kwargs: object) -> SimpleNamespace: + observed.append(argv) + return SimpleNamespace(returncode=0, stdout=json.dumps(pages), stderr="") + + monkeypatch.setattr(subprocess, "run", run) + comments = read_issue_comments( + "QuantStrategyLab/UsEquitySnapshotPipelines", + "https://github.com/QuantStrategyLab/UsEquitySnapshotPipelines/issues/123", + ) + assert len(comments) == 101 + assert "--paginate" in observed[0] and "--slurp" in observed[0] + + +def test_multiline_comment_uses_body_file_stdin(monkeypatch: pytest.MonkeyPatch) -> None: + observed: dict[str, object] = {} + + def run(argv: list[str], **kwargs: object) -> SimpleNamespace: + observed.update(argv=argv, **kwargs) + return SimpleNamespace(returncode=0, stdout="comment-url", stderr="") + + monkeypatch.setattr(subprocess, "run", run) + assert write_issue_comment( + "QuantStrategyLab/UsEquitySnapshotPipelines", + "https://github.com/QuantStrategyLab/UsEquitySnapshotPipelines/issues/123", + "line one\nline two", + ) == "comment-url" + assert observed["argv"][-2:] == ["--body-file", "-"] + assert observed["input"] == "line one\nline two" + + +def test_known_pre_numeric_failure_writes_terminal_and_prevents_future_execution(paths: dict[str, Path]) -> None: + watcher, diagnosis = watcher_inputs() + written: list[str] = [] + result = record_watcher_pre_numeric_failure( + watcher_result=watcher, diagnosis_result=diagnosis, github_app_id="42", + read_comments=lambda *_: [trusted_diagnosis_comment(watcher)], write_comment=lambda _repo, _url, body: written.append(body) or "comment", + ) + assert result == {"status": "failed", "failure_stage": "pre_numeric_failed"} + assert len(written) == 1 and ":terminal:" in written[0] + task = watcher["research_task_source_snapshot"]["tasks"][0] + diagnosis_marker = __import__("service.research_diagnosis", fromlist=["marker_for_research_diagnosis"]).marker_for_research_diagnosis(__import__("service.research_diagnosis", fromlist=["build_research_diagnosis_request"]).build_research_diagnosis_request(task)) + calls: list[list[str]] = [] + artifact = run_watcher_learning( + watcher_result=watcher, diagnosis_result={"status": "skipped", "diagnoses": []}, manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], github_app_id="42", + read_comments=lambda *_: [trusted_comment(diagnosis_marker), trusted_comment(written[0])], + write_comment=lambda *_: "comment", command_runner=lambda argv: calls.append(argv), + ) + assert artifact["failure_stage"] == "pre_numeric_failed" + assert artifact["numeric_execution"] == {"status": "reused"} + assert calls == [] + + +def test_malformed_exact_terminal_is_rejected_without_replay(paths: dict[str, Path]) -> None: + watcher, _ = watcher_inputs() + task = watcher["research_task_source_snapshot"]["tasks"][0] + diagnosis_marker = __import__("service.research_diagnosis", fromlist=["marker_for_research_diagnosis"]).marker_for_research_diagnosis(__import__("service.research_diagnosis", fromlist=["build_research_diagnosis_request"]).build_research_diagnosis_request(task)) + prefix = __import__("scripts.run_soxl_manual_learning", fromlist=["_marker"])._marker(task, "terminal") + calls: list[list[str]] = [] + artifact = run_watcher_learning( + watcher_result=watcher, diagnosis_result={"status": "skipped", "diagnoses": []}, manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], github_app_id="42", + read_comments=lambda *_: [trusted_comment(diagnosis_marker), trusted_comment(f" malformed")], + write_comment=lambda *_: "comment", command_runner=lambda argv: calls.append(argv), + ) + assert artifact["status"] == "parked" + assert artifact["failure_stage"] == "watcher_terminal_invalid" + assert calls == [] + + +@pytest.mark.parametrize( + "url", + [ + "https://example.com/QuantStrategyLab/UsEquitySnapshotPipelines/issues/1", + "https://github.com/QuantStrategyLab/OtherRepository/issues/1", + "https://github.com/QuantStrategyLab/UsEquitySnapshotPipelines/pull/1", + ], +) +def test_watcher_learning_rejects_issue_outside_exact_source_repository(url: str, paths: dict[str, Path]) -> None: + watcher, diagnosis = watcher_inputs() + watcher["issues"][0]["url"] = url + diagnosis["diagnoses"][0]["issue_url"] = url + calls: list[list[str]] = [] + artifact = run_watcher_learning( + watcher_result=watcher, diagnosis_result=diagnosis, manifest_sha256="0" * 64, + root=paths["root"], consumer_source=paths["consumer"], ues_source=paths["ues"], github_app_id="42", + read_comments=lambda *_: [], write_comment=lambda *_: "comment", + command_runner=lambda argv: calls.append(argv), + ) + assert artifact["status"] == "parked" + assert artifact["failure_stage"] == "watcher_issue_invalid" + assert calls == [] + + def test_parameter_grid_requires_unique_bounded_baseline() -> None: assert parse_parameter_grid("0.65,0.60,0.55") == (0.65, 0.6, 0.55) for invalid in ("0.60", "0.65,0.65", "0.65,0.60,0.55,0.50", "0.65,nan", "0.70,0.65"): diff --git a/tests/test_strategy_optimization_watcher_workflow.py b/tests/test_strategy_optimization_watcher_workflow.py index b2bc1cef..3573a975 100644 --- a/tests/test_strategy_optimization_watcher_workflow.py +++ b/tests/test_strategy_optimization_watcher_workflow.py @@ -80,6 +80,21 @@ def test_research_task_index_is_scheduled_and_uses_a_dedicated_sync_identity(sel self.assertIn("RESEARCH_TASK_SYNC_STATUS=NOT_CONFIGURED", text) self.assertNotIn("/api/switch", text) + def test_exact_soxl_task_hands_off_only_sanitized_artifacts_to_vps_consumer(self) -> None: + text = WORKFLOW_PATH.read_text(encoding="utf-8") + self.assertIn("soxl-watcher-learning-ready", text) + self.assertIn("needs.strategy-optimization-watcher.outputs.soxl_learning_ready == 'true'", text) + self.assertIn("vars.SOXL_WATCHER_GCP_WIF_PROVIDER != ''", text) + self.assertIn("vars.SOXL_WATCHER_GCP_PROJECT_ID != ''", text) + self.assertLess(text.index("vars.SOXL_WATCHER_GCP_WIF_PROVIDER != ''"), text.index("runs-on: [self-hosted, codex-vps]")) + self.assertIn("runs-on: [self-hosted, codex-vps]", text) + self.assertIn("python -m scripts.run_soxl_manual_learning --watcher-result", text) + self.assertIn("strategy-optimization-watcher-${{ github.run_id }}", text) + self.assertIn("GH_TOKEN: ${{ steps.source_app_token.outputs.token || github.token }}", text) + self.assertIn("soxl-p1-p3/${P1_MANIFEST_SHA256}", text) + self.assertIn("--body-file", Path(__file__).resolve().parents[1].joinpath("scripts/run_soxl_manual_learning.py").read_text()) + self.assertNotIn("bars.json=${{ needs.", text) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_strategy_watch.py b/tests/test_strategy_watch.py index 78caa928..552d0fad 100644 --- a/tests/test_strategy_watch.py +++ b/tests/test_strategy_watch.py @@ -13,10 +13,52 @@ resolve_strategy_watch_repository, watcher_issue_key, ) +from service.research_task import SOXL_WATCHER_PARAMETER_BOUNDS_SHA256, validate_strategy_diagnosis_task class StrategyWatchTest(unittest.TestCase): + def test_exact_soxl_v3_task_binds_the_fixed_learning_experiment(self) -> None: + finding = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/UsEquitySnapshotPipelines", + "strategy_profile": "soxl_soxx_core_only_p2_v3", + "candidate_kind": "individual", + "domain": "us_equity", + "schema_version": "strategy_performance.v2", + "metrics_kind": "performance", + "generated_at": "2026-09-10T00:00:00Z", + "current_metrics": {"sharpe": 0.5, "cagr": 0.1, "calmar": 0.7, "win_rate": 0.52, "max_dd": 0.12}, + "baseline_metrics": {"sharpe": 1.0, "cagr": 0.2, "calmar": 1.0, "win_rate": 0.58, "max_dd": 0.08}, + "research_task_evidence": { + "p1_input_digest": "a" * 64, + "p2_config_digest": "ff8fa0acf4f175a7c40c3e1e6a3304ea2748b6b81c3797342085a4df3810ab4d", + "p3_evidence_id": "c" * 64, + "strategy_revision": "7756fe32585e85cf1d09a163203a02e3eee39fe1", + "producer_revision": "e" * 40, + }, + } + )[0] + + task = validate_strategy_diagnosis_task(__import__("service.strategy_watch", fromlist=["finding_to_research_task"]).finding_to_research_task(finding)) + repeated = validate_strategy_diagnosis_task(__import__("service.strategy_watch", fromlist=["finding_to_research_task"]).finding_to_research_task(finding)) + self.assertEqual(task["experiment"]["parameter_bounds_sha256"], SOXL_WATCHER_PARAMETER_BOUNDS_SHA256) + self.assertEqual(task["task_id"], repeated["task_id"]) + self.assertEqual(task["task_sha256"], repeated["task_sha256"]) + + def test_old_null_bounds_soxl_task_remains_valid_but_non_executable(self) -> None: + from service.research_task import build_strategy_diagnosis_task, calculate_task_sha256 + + task = build_strategy_diagnosis_task( + event_key="123456789abc", created_at="2026-09-10T00:00:00Z", + candidate_id="soxl_soxx_core_only_p2_v3", candidate_kind="individual", + domain="us_equity", strategy_repository="QuantStrategyLab/UsEquityStrategies", + evidence={"p1_input_digest": "a" * 64, "p2_config_digest": "ff8fa0acf4f175a7c40c3e1e6a3304ea2748b6b81c3797342085a4df3810ab4d", "p3_evidence_id": "c" * 64, "strategy_revision": "7756fe32585e85cf1d09a163203a02e3eee39fe1", "producer_revision": "e" * 40}, + ) + task["experiment"]["parameter_bounds_sha256"] = None + task["task_sha256"] = calculate_task_sha256(task) + self.assertIsNone(validate_strategy_diagnosis_task(task)["experiment"]["parameter_bounds_sha256"]) + def test_deferred_research_input_creates_issue_only_data_finding(self) -> None: finding = build_research_input_unavailable_finding( repo="QuantStrategyLab/UsEquitySnapshotPipelines",