Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions .github/workflows/strategy_optimization_watcher.yml
Original file line number Diff line number Diff line change
Expand Up @@ -272,7 +272,7 @@ jobs:
| tee data/output/strategy_optimization_watcher/research-diagnosis.json

- name: Extract bounded research task source snapshot
if: steps.fetch-metrics.outputs.downloaded == 'true'
if: success()
working-directory: bridge
run: |
set -euo pipefail
Expand All @@ -289,7 +289,7 @@ jobs:
PY

- name: Publish research task index to the unified console
if: steps.fetch-metrics.outputs.downloaded == 'true' && github.event_name == 'schedule'
if: success() && github.event_name == 'schedule'
env:
RESEARCH_TASK_SYNC_URL: ${{ vars.QSL_RESEARCH_TASK_SYNC_URL }}
RESEARCH_TASK_SYNC_TOKEN: ${{ secrets.QSL_RESEARCH_TASK_SYNC_TOKEN }}
Expand Down
49 changes: 44 additions & 5 deletions scripts/run_strategy_optimization_watcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -304,13 +304,24 @@ def run_research_input_terminal_watcher(
comment_issue: Callable[[str, str, str], str] = comment_github_issue,
list_issues: Callable[[str], dict[str, str]] = list_open_issue_urls,
) -> dict[str, Any]:
"""Surface a trusted deferred P1 record as an issue-only finding."""
"""Surface a trusted deferred P1 record as an issue-only finding.

An accepted P1 terminal record is not a failure. It simply means that
the producer has not yet emitted the two comparable P3 observations the
watcher needs. Keep that state visible to the unified console without
opening a misleading issue or failing the scheduled watcher.
"""
candidate = terminal.get("candidate") if isinstance(terminal.get("candidate"), dict) else {}
status = str(terminal.get("status") or "").strip().upper()
reason_code = str(terminal.get("reason_code") or "").strip()
if status != "DEFERRED" or not reason_code:
reason = "p1_terminal_accepted" if status == "ACCEPTED" else "p1_terminal_contract_unavailable"
return no_comparable_metrics_result(reason=reason, dry_run=dry_run)
finding = build_research_input_unavailable_finding(
repo=source_repo,
profile=profile,
status=str(terminal.get("status") or ""),
reason_code=str(terminal.get("reason_code") or ""),
status=status,
reason_code=reason_code,
candidate_id=str(candidate.get("candidate_id") or ""),
date_cutoff=str(terminal.get("date_cutoff") or ""),
source=source,
Expand All @@ -331,6 +342,32 @@ def run_research_input_terminal_watcher(
return result


def no_comparable_metrics_result(
*, reason: str = "comparable_metrics_unavailable", dry_run: bool = True
) -> dict[str, Any]:
"""Return a successful, source-owned unavailable queue snapshot.

This is deliberately not an exception: optimization requires two trusted
comparable P3 observations. Until they exist, the console must show an
unavailable source rather than silently retaining stale tasks or marking
an accepted P1 acquisition as a watcher failure.
"""
snapshot = research_task_source_snapshot(
[],
context_available=False,
computed_at=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
)
snapshot["errors"] = sorted(set(snapshot["errors"] + [reason]))
return {
"status": "ok",
"dry_run": dry_run,
"findings": 0,
"issues": [],
"errors": 0,
"research_task_source_snapshot": snapshot,
}


def main() -> int:
try:
input_path = resolve_input_path(
Expand All @@ -342,7 +379,8 @@ def main() -> int:
print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True))
return 2
if input_path is None:
print(json.dumps({"status": "skipped", "reason": "strategy metrics input not configured"}, sort_keys=True))
result = no_comparable_metrics_result(reason="metrics_input_not_configured")
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
return 0
terminal_path_text = os.environ.get("STRATEGY_WATCH_TERMINAL_STATUS_PATH", "").strip()
terminal_path = None
Expand All @@ -357,7 +395,8 @@ def main() -> int:
return 2
if not input_path.exists():
if terminal_path is None or not terminal_path.is_file():
print(json.dumps({"status": "skipped", "reason": "strategy metrics input not found — this is expected when the source repository has not yet published metrics"}, sort_keys=True))
result = no_comparable_metrics_result()
print(json.dumps(result, ensure_ascii=False, sort_keys=True))
return 0
try:
terminal = load_payload(terminal_path)
Expand Down
50 changes: 50 additions & 0 deletions tests/test_run_strategy_optimization_watcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
import subprocess
import tempfile
import unittest
from contextlib import redirect_stdout
from io import StringIO
from unittest.mock import patch

from scripts.run_strategy_optimization_watcher import (
Expand Down Expand Up @@ -103,6 +105,54 @@ def test_deferred_terminal_creates_issue_only_finding(self) -> None:
self.assertEqual(result["findings"], 1)
self.assertEqual(len(created), 1)
self.assertEqual(result["issues"][0]["task"]["trigger"]["kind"], "strategy_research_input_unavailable")

def test_accepted_terminal_is_visible_but_not_a_watcher_failure(self) -> None:
result = run_research_input_terminal_watcher(
{
"status": "ACCEPTED",
"reason_code": "",
"candidate": {"candidate_id": "soxl_soxx_core_only_p2_v3"},
},
source_repo="QuantStrategyLab/UsEquitySnapshotPipelines",
profile="soxl_soxx_trend_income",
dry_run=False,
)

snapshot = result["research_task_source_snapshot"]
self.assertEqual(result["status"], "ok")
self.assertFalse(result["dry_run"])
self.assertEqual(result["errors"], 0)
self.assertEqual(result["findings"], 0)
self.assertEqual(snapshot["data_status"], "unavailable")
self.assertEqual(snapshot["tasks"], [])
self.assertIn("p1_terminal_accepted", snapshot["errors"])
self.assertIn("research_task_context_unavailable", snapshot["errors"])

def test_main_publishes_unavailable_snapshot_when_metrics_do_not_exist(self) -> None:
with tempfile.TemporaryDirectory() as directory:
original = dict(os.environ)
output = StringIO()
try:
os.environ.update(
{
"STRATEGY_WATCH_SOURCE_ROOT": directory,
"STRATEGY_WATCH_METRICS_PATH": "data/output/not-yet-published.json",
"STRATEGY_WATCH_SOURCE_REPO": "QuantStrategyLab/UsEquitySnapshotPipelines",
"STRATEGY_WATCH_DRY_RUN": "true",
}
)
with redirect_stdout(output):
self.assertEqual(main(), 0)
finally:
os.environ.clear()
os.environ.update(original)

result = json.loads(output.getvalue())
snapshot = result["research_task_source_snapshot"]
self.assertEqual(result["status"], "ok")
self.assertEqual(snapshot["data_status"], "unavailable")
self.assertIn("comparable_metrics_unavailable", snapshot["errors"])

def test_monitoring_dispatch_does_not_repeat_existing_issue(self) -> None:
finding = build_strategy_monitoring_finding(
domain="crypto",
Expand Down