From 76503d0ed33ae43a3a73698ea0d56fc87568ce71 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 01:26:55 +0800 Subject: [PATCH 1/9] feat: add strategy optimization watcher Co-Authored-By: Codex --- .../strategy_optimization_watcher.yml | 125 ++++++++++++ docs/ai_autonomy_architecture.md | 22 +++ scripts/run_strategy_optimization_watcher.py | 169 ++++++++++++++++ service/automation_contracts.py | 122 ++++++++++++ service/strategy_optimization_policy.py | 130 +++++++++++++ service/strategy_watch.py | 182 ++++++++++++++++++ tests/test_automation_contracts.py | 176 +++++++++++++++++ .../test_run_strategy_optimization_watcher.py | 88 +++++++++ tests/test_strategy_optimization_policy.py | 44 +++++ ..._strategy_optimization_watcher_workflow.py | 43 +++++ tests/test_strategy_watch.py | 64 ++++++ 11 files changed, 1165 insertions(+) create mode 100644 .github/workflows/strategy_optimization_watcher.yml create mode 100755 scripts/run_strategy_optimization_watcher.py create mode 100644 service/automation_contracts.py create mode 100644 service/strategy_optimization_policy.py create mode 100644 service/strategy_watch.py create mode 100644 tests/test_automation_contracts.py create mode 100644 tests/test_run_strategy_optimization_watcher.py create mode 100644 tests/test_strategy_optimization_policy.py create mode 100644 tests/test_strategy_optimization_watcher_workflow.py create mode 100644 tests/test_strategy_watch.py diff --git a/.github/workflows/strategy_optimization_watcher.yml b/.github/workflows/strategy_optimization_watcher.yml new file mode 100644 index 00000000..ca711cb6 --- /dev/null +++ b/.github/workflows/strategy_optimization_watcher.yml @@ -0,0 +1,125 @@ +name: Strategy Optimization Watcher + +on: + workflow_dispatch: + inputs: + source_repo: + description: "Repository that owns strategy metrics and receives optimization issues" + required: true + default: "QuantStrategyLab/CryptoLivePoolPipelines" + source_ref: + description: "Source repository ref to inspect" + required: false + default: "main" + metrics_path: + description: "JSON metrics payload path inside the source repository" + required: false + default: "data/output/strategy_metrics.json" + dry_run: + description: "Do not create GitHub issues" + required: false + type: boolean + default: true + schedule: + - cron: "17 3 * * *" + +permissions: + contents: read + issues: write + +concurrency: + group: strategy-optimization-watcher-${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || github.repository }} + cancel-in-progress: false + +jobs: + strategy-optimization-watcher: + runs-on: ubuntu-latest + timeout-minutes: 15 + env: + FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" + SOURCE_REPO: ${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || github.repository }} + SOURCE_REF: ${{ github.event.inputs.source_ref || vars.STRATEGY_WATCH_SOURCE_REF || 'main' }} + METRICS_PATH: ${{ github.event.inputs.metrics_path || vars.STRATEGY_WATCH_METRICS_PATH || 'data/output/strategy_metrics.json' }} + STRATEGY_WATCH_DRY_RUN: ${{ github.event.inputs.dry_run || vars.STRATEGY_WATCH_DRY_RUN || 'true' }} + steps: + - name: Checkout Bridge + uses: actions/checkout@v6.0.3 + with: + path: bridge + persist-credentials: false + + - name: Detect GitHub App Credentials + id: app_credentials + env: + APP_ID: ${{ vars.CROSS_REPO_GITHUB_APP_ID }} + APP_PRIVATE_KEY: ${{ secrets.CROSS_REPO_GITHUB_APP_PRIVATE_KEY }} + run: | + set -euo pipefail + if [ -n "${APP_ID:-}" ] && [ -n "${APP_PRIVATE_KEY:-}" ]; then + echo "available=true" >> "$GITHUB_OUTPUT" + else + echo "available=false" >> "$GITHUB_OUTPUT" + fi + + - name: Resolve Source Repository Name + id: source_repo + run: | + set -euo pipefail + if [[ ! "${SOURCE_REPO}" =~ ^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$ ]]; then + echo "Invalid SOURCE_REPO: ${SOURCE_REPO}. Expected owner/name." >&2 + exit 1 + fi + repository="${SOURCE_REPO#*/}" + echo "repository=${repository}" >> "$GITHUB_OUTPUT" + + - name: Create GitHub App Token For Source Repository + id: source_app_token + if: steps.app_credentials.outputs.available == 'true' + continue-on-error: 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: ${{ github.repository_owner }} + repositories: ${{ steps.source_repo.outputs.repository }} + permission-contents: read + permission-issues: write + + - name: Verify Source Repository Token + env: + SOURCE_APP_TOKEN: ${{ steps.source_app_token.outputs.token }} + run: | + set -euo pipefail + if [ "${SOURCE_REPO}" != "${GITHUB_REPOSITORY}" ] && [ -z "${SOURCE_APP_TOKEN:-}" ]; then + echo "Cross-repository strategy watcher requires CROSS_REPO_GITHUB_APP_ID and CROSS_REPO_GITHUB_APP_PRIVATE_KEY." >&2 + exit 1 + fi + + - name: Checkout Source Metrics + uses: actions/checkout@v6.0.3 + with: + repository: ${{ env.SOURCE_REPO }} + ref: ${{ env.SOURCE_REF }} + path: source + token: ${{ steps.source_app_token.outputs.token || github.token }} + persist-credentials: false + + - name: Run Strategy Optimization Watcher + env: + GH_TOKEN: ${{ steps.source_app_token.outputs.token || github.token }} + STRATEGY_WATCH_SOURCE_ROOT: ${{ github.workspace }}/source + STRATEGY_WATCH_METRICS_PATH: ${{ env.METRICS_PATH }} + STRATEGY_WATCH_SOURCE_REPO: ${{ env.SOURCE_REPO }} + working-directory: bridge + run: | + set -euo pipefail + mkdir -p data/output/strategy_optimization_watcher + python scripts/run_strategy_optimization_watcher.py | tee data/output/strategy_optimization_watcher/result.json + + - name: Upload watcher diagnostics + if: always() + uses: actions/upload-artifact@v7 + with: + name: strategy-optimization-watcher-${{ github.run_id }} + path: bridge/data/output/strategy_optimization_watcher/ + if-no-files-found: warn diff --git a/docs/ai_autonomy_architecture.md b/docs/ai_autonomy_architecture.md index 92265815..41902644 100644 --- a/docs/ai_autonomy_architecture.md +++ b/docs/ai_autonomy_architecture.md @@ -257,6 +257,28 @@ AIAuditBridge 是 QuantStrategyLab 的 AI 审计控制面，负责： 这类优化有价值，但不是无人值守的先决条件。 +#### 3.8 Strategy Optimization Watcher 的 issue-only 安全边界 + +对于 Strategy Optimization Watcher，首批实现只走 issue-only 提案流，不直接触碰策略执行面。 + +推荐流程是： + +1. deterministic trigger 触发 watcher； +2. 生成 evidence bundle； +3. 只创建 optimization issue / task proposal； +4. 经过 authority / registry gate 校验后再决定是否进入下一步； +5. 后续如需执行，再由人工或 CI gate 接管。 + +当前首批实现的安全边界是： + +- 只创建 optimization issue/task； +- 不自动改策略； +- 不调 live 参数； +- 不联网检索； +- 不自动 merge / deploy。 + +这样可以把策略优化先收敛为可审计、可回放的建议流，再逐步扩展到受控执行面。 + --- ## 4. 可落地的阶段性改造计划 diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py new file mode 100755 index 00000000..5e492182 --- /dev/null +++ b/scripts/run_strategy_optimization_watcher.py @@ -0,0 +1,169 @@ +#!/usr/bin/env python3 +"""Run the issue-only strategy optimization watcher.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path, PurePosixPath +import re +import subprocess +import sys +from typing import Any, Callable + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +from service.strategy_watch import evaluate_strategy_watch, finding_to_automation_task, issue_for_task # noqa: E402 + +REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") + + +def parse_bool(value: Any, *, default: bool = False) -> bool: + if value is None: + return default + if isinstance(value, bool): + return value + text = str(value).strip().lower() + if not text: + return default + return text in {"1", "true", "yes", "on"} + + +def resolve_input_path( + *, + input_path: str = "", + source_root: str = "", + metrics_path: str = "", +) -> Path | None: + if source_root and metrics_path: + normalized = PurePosixPath(metrics_path.replace("\\", "/")) + if normalized.is_absolute() or ".." in normalized.parts: + raise ValueError("metrics_path must be a relative path inside the source checkout") + root = Path(source_root).resolve() + candidate = (root / Path(*normalized.parts)).resolve() + try: + candidate.relative_to(root) + except ValueError as exc: + raise ValueError("metrics_path resolves outside the source checkout") from exc + return candidate + if not input_path: + return None + candidate = Path(input_path).resolve() + if source_root: + root = Path(source_root).resolve() + try: + candidate.relative_to(root) + except ValueError as exc: + raise ValueError("input path resolves outside the source checkout") from exc + return candidate + + +def load_payload(path: str | Path) -> dict[str, Any]: + payload = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(payload, dict): + raise ValueError("strategy watch input must be a JSON object") + return payload + + +def find_existing_open_issue(repo: str, title: str) -> str: + if not REPO_RE.fullmatch(repo): + raise ValueError("repository must be in owner/name form") + result = subprocess.run( + ["gh", "issue", "list", "--repo", repo, "--state", "open", "--limit", "1000", "--json", "title,url"], + check=True, + capture_output=True, + text=True, + ) + try: + issues = json.loads(result.stdout or "[]") + except json.JSONDecodeError: + return "" + if not isinstance(issues, list): + return "" + for issue in issues: + if isinstance(issue, dict) and issue.get("title") == title: + return str(issue.get("url") or "") + return "" + + +def create_github_issue(repo: str, title: str, body: str) -> str: + if not REPO_RE.fullmatch(repo): + raise ValueError("repository must be in owner/name form") + result = subprocess.run( + ["gh", "issue", "create", "--repo", repo, "--title", title, "--body", body], + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + +def run_watcher( + payload: dict[str, Any], + *, + source_repo: str = "", + dry_run: bool = True, + create_issue: Callable[[str, str, str], str] = create_github_issue, + find_issue: Callable[[str, str], str] = find_existing_open_issue, +) -> dict[str, Any]: + findings = evaluate_strategy_watch(payload) + issues: list[dict[str, Any]] = [] + for finding in findings: + task = finding_to_automation_task(finding) + issue = issue_for_task(task) + repo = source_repo or finding.snapshot.repo + issue_result: dict[str, Any] = { + "repo": repo, + "title": issue["title"], + "task": task.to_dict(), + "created": False, + } + if dry_run: + issue_result["dry_run"] = True + else: + existing_url = find_issue(repo, issue["title"]) + if existing_url: + issue_result["existing_url"] = existing_url + issue_result["skipped_reason"] = "open issue already exists" + else: + issue_result["url"] = create_issue(repo, issue["title"], issue["body"]) + issue_result["created"] = True + issues.append(issue_result) + return { + "status": "ok", + "dry_run": dry_run, + "findings": len(findings), + "issues": issues, + } + + +def main() -> int: + try: + input_path = resolve_input_path( + input_path=os.environ.get("STRATEGY_WATCH_INPUT", "").strip(), + source_root=os.environ.get("STRATEGY_WATCH_SOURCE_ROOT", "").strip(), + metrics_path=os.environ.get("STRATEGY_WATCH_METRICS_PATH", "").strip(), + ) + except ValueError as exc: + 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)) + return 0 + if not input_path.exists(): + print(json.dumps({"status": "error", "error": "strategy metrics input not found"}, sort_keys=True)) + return 2 + payload = load_payload(input_path) + result = run_watcher( + payload, + source_repo=os.environ.get("STRATEGY_WATCH_SOURCE_REPO", "").strip(), + dry_run=parse_bool(os.environ.get("STRATEGY_WATCH_DRY_RUN"), default=True), + ) + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/service/automation_contracts.py b/service/automation_contracts.py new file mode 100644 index 00000000..35aa492d --- /dev/null +++ b/service/automation_contracts.py @@ -0,0 +1,122 @@ +"""Minimal shared contracts for automation strategy data.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + + +def _validate_non_empty(value: str, field_name: str) -> None: + if not str(value).strip(): + raise ValueError(f"{field_name} must be a non-empty string") + + +@dataclass +class TriggerRecord: + source: str + kind: str + severity: str + reason: str + subject: str + metrics: dict[str, Any] = field(default_factory=dict) + evidence: list[Any] = field(default_factory=list) + created_at: float | None = None + + def __post_init__(self) -> None: + _validate_non_empty(self.severity, "severity") + + def to_dict(self) -> dict[str, Any]: + return { + "source": self.source, + "kind": self.kind, + "severity": self.severity, + "reason": self.reason, + "subject": self.subject, + "metrics": dict(self.metrics), + "evidence": list(self.evidence), + "created_at": self.created_at, + } + + +@dataclass +class EvidenceBundle: + summary: str + artifacts: list[Any] = field(default_factory=list) + metrics: dict[str, Any] = field(default_factory=dict) + risks: list[Any] = field(default_factory=list) + + def to_dict(self) -> dict[str, Any]: + return { + "summary": self.summary, + "artifacts": list(self.artifacts), + "metrics": dict(self.metrics), + "risks": list(self.risks), + } + + +@dataclass +class ProposedAction: + action: str + lane: str + target: str + rationale: str + requires_human_review: bool = True + metadata: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + _validate_non_empty(self.action, "action") + + def to_dict(self) -> dict[str, Any]: + return { + "action": self.action, + "lane": self.lane, + "target": self.target, + "rationale": self.rationale, + "requires_human_review": self.requires_human_review, + "metadata": dict(self.metadata), + } + + +@dataclass +class GateDecision: + allowed: bool + reason: str + required_checks: list[Any] = field(default_factory=list) + human_review_required: bool = True + metadata: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "allowed": self.allowed, + "reason": self.reason, + "required_checks": list(self.required_checks), + "human_review_required": self.human_review_required, + "metadata": dict(self.metadata), + } + + +@dataclass +class AutomationTask: + trigger: TriggerRecord + evidence: EvidenceBundle + proposed_action: ProposedAction + gate_decision: GateDecision + status: str = "proposed" + metadata: dict[str, Any] = field(default_factory=dict) + + def __post_init__(self) -> None: + _validate_non_empty(self.status, "status") + + @property + def is_actionable(self) -> bool: + return self.gate_decision.allowed and bool(self.proposed_action.action.strip()) + + def to_dict(self) -> dict[str, Any]: + return { + "trigger": self.trigger.to_dict(), + "evidence": self.evidence.to_dict(), + "proposed_action": self.proposed_action.to_dict(), + "gate_decision": self.gate_decision.to_dict(), + "status": self.status, + "metadata": dict(self.metadata), + } diff --git a/service/strategy_optimization_policy.py b/service/strategy_optimization_policy.py new file mode 100644 index 00000000..455c82cd --- /dev/null +++ b/service/strategy_optimization_policy.py @@ -0,0 +1,130 @@ +"""Deterministic policy for strategy optimization watch triggers.""" + +from __future__ import annotations + +from copy import deepcopy +from dataclasses import dataclass, field +from typing import Any + +SEVERITY_NONE = "none" +SEVERITY_MEDIUM = "medium" +SEVERITY_HIGH = "high" + +DEFAULT_METRIC_RULES: dict[str, dict[str, Any]] = { + "sharpe": {"higher_better": True, "relative_drop": 0.05}, + "cagr": {"higher_better": True, "relative_drop": 0.05}, + "calmar": {"higher_better": True, "relative_drop": 0.05}, + "win_rate": {"higher_better": True, "relative_drop": 0.03}, + "max_dd": {"higher_better": False, "absolute_worsening": 0.02}, +} +HIGH_SEVERITY_SIGNAL_COUNT = 2 +HIGH_SEVERITY_MAX_DD_WORSENING = 0.05 + + +@dataclass(frozen=True) +class StrategyOptimizationPolicy: + """Thresholds used by the watcher; deterministic and service-owned.""" + + metric_rules: dict[str, dict[str, Any]] = field(default_factory=lambda: deepcopy(DEFAULT_METRIC_RULES)) + high_severity_signal_count: int = HIGH_SEVERITY_SIGNAL_COUNT + high_severity_max_dd_worsening: float = HIGH_SEVERITY_MAX_DD_WORSENING + + +def _safe_float(value: Any) -> float | None: + if isinstance(value, bool): + return None + try: + return float(value) + except (TypeError, ValueError): + return None + + +def metric_degradation_signals( + current_metrics: dict[str, Any], + baseline_metrics: dict[str, Any], + *, + policy: StrategyOptimizationPolicy | None = None, +) -> list[dict[str, Any]]: + """Return deterministic degradation signals beyond configured thresholds.""" + active_policy = policy or StrategyOptimizationPolicy() + signals: list[dict[str, Any]] = [] + for metric, rule in active_policy.metric_rules.items(): + current = _safe_float(current_metrics.get(metric)) + baseline = _safe_float(baseline_metrics.get(metric)) + if current is None or baseline is None: + continue + delta = current - baseline + signal: dict[str, Any] | None = None + if rule.get("higher_better") is False: + threshold = float(rule.get("absolute_worsening", 0.0)) + if delta > threshold: + signal = { + "metric": metric, + "baseline": baseline, + "current": current, + "delta": delta, + "threshold": threshold, + "reason": f"{metric} worsened by {delta:.4g} > {threshold:.4g}", + } + else: + threshold = float(rule.get("relative_drop", 0.0)) + absolute_threshold = float(rule.get("absolute_drop_when_zero", threshold)) + if baseline == 0: + relative_delta = None + degraded = current < -absolute_threshold + else: + relative_delta = delta / abs(baseline) + degraded = relative_delta < -threshold + if degraded: + reason = ( + f"{metric} dropped below zero by {abs(current):.4g} > {absolute_threshold:.4g}" + if relative_delta is None + else f"{metric} dropped {relative_delta:.1%} beyond {threshold:.1%}" + ) + signal = { + "metric": metric, + "baseline": baseline, + "current": current, + "delta": delta, + "relative_delta": relative_delta, + "threshold": threshold, + "reason": reason, + } + if signal is not None: + signals.append(signal) + return signals + + +def classify_strategy_degradation( + signals: list[dict[str, Any]], + *, + policy: StrategyOptimizationPolicy | None = None, +) -> str: + """Classify watcher severity without involving an LLM.""" + if not signals: + return SEVERITY_NONE + active_policy = policy or StrategyOptimizationPolicy() + if len(signals) >= active_policy.high_severity_signal_count: + return SEVERITY_HIGH + for signal in signals: + if signal.get("metric") == "max_dd" and float(signal.get("delta") or 0.0) >= active_policy.high_severity_max_dd_worsening: + return SEVERITY_HIGH + return SEVERITY_MEDIUM + + +def evaluate_strategy_metrics( + current_metrics: dict[str, Any], + baseline_metrics: dict[str, Any], + *, + policy: StrategyOptimizationPolicy | None = None, +) -> dict[str, Any]: + """Evaluate whether a strategy profile should open an optimization issue.""" + active_policy = policy or StrategyOptimizationPolicy() + signals = metric_degradation_signals(current_metrics, baseline_metrics, policy=active_policy) + severity = classify_strategy_degradation(signals, policy=active_policy) + return { + "should_open_issue": bool(signals), + "severity": severity, + "signals": signals, + "signal_count": len(signals), + } diff --git a/service/strategy_watch.py b/service/strategy_watch.py new file mode 100644 index 00000000..631c6235 --- /dev/null +++ b/service/strategy_watch.py @@ -0,0 +1,182 @@ +"""Strategy optimization watcher for issue-only automation proposals.""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any + +from service.automation_contracts import AutomationTask, EvidenceBundle, GateDecision, ProposedAction, TriggerRecord +from service.strategy_automation_registry import LANE_RESEARCH_BACKLOG, summarize_strategy_registry_context +from service.strategy_optimization_policy import evaluate_strategy_metrics + +WATCHER_SCHEMA_VERSION = "strategy_optimization_watch.v1" +ISSUE_ONLY_ACTION = "open_issue" + + +@dataclass(frozen=True) +class StrategyWatchSnapshot: + repo: str + profile: str + plugin: str = "" + current_metrics: dict[str, Any] = field(default_factory=dict) + baseline_metrics: dict[str, Any] = field(default_factory=dict) + source: str = "" + generated_at: str = "" + + @classmethod + def from_dict(cls, payload: dict[str, Any], *, default_repo: str = "") -> "StrategyWatchSnapshot": + profile = str(payload.get("strategy_profile") or payload.get("profile") or "").strip() + return cls( + repo=str(payload.get("repo") or payload.get("repository") or default_repo).strip(), + profile=profile, + plugin=str(payload.get("plugin") or payload.get("strategy_plugin") or "").strip(), + current_metrics=dict(payload.get("current_metrics") or payload.get("current") or {}), + baseline_metrics=dict(payload.get("baseline_metrics") or payload.get("baseline") or {}), + source=str(payload.get("source") or "").strip(), + generated_at=str(payload.get("generated_at") or "").strip(), + ) + + def subject(self) -> str: + parts = [self.repo, self.profile or self.plugin] + return ":".join(part for part in parts if part) + + def to_dict(self) -> dict[str, Any]: + return { + "repo": self.repo, + "profile": self.profile, + "plugin": self.plugin, + "current_metrics": self.current_metrics, + "baseline_metrics": self.baseline_metrics, + "source": self.source, + "generated_at": self.generated_at, + } + + +@dataclass(frozen=True) +class StrategyWatchFinding: + snapshot: StrategyWatchSnapshot + severity: str + signals: list[dict[str, Any]] + registry_context: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "schema_version": WATCHER_SCHEMA_VERSION, + "snapshot": self.snapshot.to_dict(), + "severity": self.severity, + "signals": self.signals, + "registry_context": self.registry_context, + } + + +def _snapshots_from_payload(payload: dict[str, Any]) -> list[StrategyWatchSnapshot]: + default_repo = str(payload.get("repo") or payload.get("repository") or "").strip() + raw_snapshots = payload.get("snapshots") + if not isinstance(raw_snapshots, list): + raw_snapshots = [payload] + snapshots: list[StrategyWatchSnapshot] = [] + for item in raw_snapshots: + if isinstance(item, dict): + snapshots.append(StrategyWatchSnapshot.from_dict(item, default_repo=default_repo)) + return snapshots + + +def evaluate_strategy_watch(payload: dict[str, Any]) -> list[StrategyWatchFinding]: + """Evaluate metrics payload and return issue-worthy findings only.""" + registry_payload = payload.get("automation_registry") or payload.get("registry") or {} + findings: list[StrategyWatchFinding] = [] + for snapshot in _snapshots_from_payload(payload): + decision = evaluate_strategy_metrics(snapshot.current_metrics, snapshot.baseline_metrics) + if not decision["should_open_issue"]: + continue + context = summarize_strategy_registry_context(registry_payload, snapshot.profile) if snapshot.profile else {} + findings.append( + StrategyWatchFinding( + snapshot=snapshot, + severity=str(decision["severity"]), + signals=list(decision["signals"]), + registry_context=context, + ) + ) + return findings + + +def finding_to_automation_task(finding: StrategyWatchFinding) -> AutomationTask: + """Convert a deterministic finding into an issue-only automation task.""" + lane = str(finding.registry_context.get("automation_lane") or LANE_RESEARCH_BACKLOG) + signal_reasons = [str(signal.get("reason") or signal.get("metric") or "metric degraded") for signal in finding.signals] + trigger = TriggerRecord( + source="strategy_optimization_watcher", + kind="strategy_metric_degradation", + severity=finding.severity, + reason="; ".join(signal_reasons) or "strategy metrics degraded", + subject=finding.snapshot.subject(), + metrics=finding.snapshot.current_metrics, + evidence=signal_reasons, + ) + evidence = EvidenceBundle( + summary="Deterministic strategy metrics crossed degradation thresholds.", + artifacts=[finding.snapshot.source] if finding.snapshot.source else [], + metrics={ + "current": finding.snapshot.current_metrics, + "baseline": finding.snapshot.baseline_metrics, + }, + risks=[ + "issue-only: no strategy code, live parameters, broker/order paths, or deployment are changed", + "sandbox backtest evidence is required before any PR can be proposed", + ], + ) + proposed = ProposedAction( + action=ISSUE_ONLY_ACTION, + lane=lane, + target=finding.snapshot.repo, + rationale="Open a research optimization issue for AI diagnosis and sandbox experiment planning.", + requires_human_review=True, + metadata={"profile": finding.snapshot.profile, "plugin": finding.snapshot.plugin}, + ) + gate = GateDecision( + allowed=True, + reason="Issue-only proposal is allowed; code changes and live-impact actions remain gated.", + required_checks=[ + "human review before strategy code changes", + "sandbox backtest before optimization PR", + "strategy registry/authority gate before live impact", + ], + human_review_required=True, + metadata={"issue_only": True, "live_impact_allowed": False}, + ) + return AutomationTask(trigger=trigger, evidence=evidence, proposed_action=proposed, gate_decision=gate) + + +def issue_for_task(task: AutomationTask) -> dict[str, str]: + """Build a GitHub issue title/body for a strategy optimization task.""" + payload = task.to_dict() + trigger = payload["trigger"] + evidence = payload["evidence"] + action = payload["proposed_action"] + title = f"AI strategy optimization proposal: {trigger.get('subject') or action.get('target') or 'strategy profile'}" + signals = "\n".join(f"- {item}" for item in trigger.get("evidence", [])) or "- Strategy metrics degraded." + checks = "\n".join(f"- [ ] {item}" for item in payload["gate_decision"].get("required_checks", [])) + risks = "\n".join(f"- {item}" for item in evidence.get("risks", [])) + body = "\n".join( + [ + "## Summary", + str(evidence.get("summary") or "Strategy optimization watcher opened this issue."), + "", + "## Trigger", + f"- Severity: `{trigger.get('severity')}`", + f"- Subject: `{trigger.get('subject')}`", + "", + "## Signals", + signals, + "", + "## Safety boundary", + risks, + "", + "## Required gates before code/live impact", + checks, + "", + "This watcher only opens an issue. It does not modify strategy code, tune live parameters, merge PRs, or deploy.", + ] + ) + return {"title": title[:240], "body": body} diff --git a/tests/test_automation_contracts.py b/tests/test_automation_contracts.py new file mode 100644 index 00000000..72409d45 --- /dev/null +++ b/tests/test_automation_contracts.py @@ -0,0 +1,176 @@ +"""Tests for service/automation_contracts.py — minimal automation data model.""" + +from __future__ import annotations + +import unittest + +from service.automation_contracts import ( + AutomationTask, + EvidenceBundle, + GateDecision, + ProposedAction, + TriggerRecord, +) + + +class TestAutomationContracts(unittest.TestCase): + def test_to_dict(self) -> None: + trigger = TriggerRecord( + source="watcher", + kind="strategy_update", + severity="high", + reason="threshold exceeded", + subject="strategy-a", + metrics={"score": 0.91}, + evidence=["log-1"], + created_at=123.4, + ) + evidence = EvidenceBundle( + summary="summary", + artifacts=["artifact-1"], + metrics={"latency": 42}, + risks=["regression"], + ) + action = ProposedAction( + action="adjust_threshold", + lane="automation", + target="strategy-a", + rationale="reduce risk", + requires_human_review=False, + metadata={"source": "test"}, + ) + decision = GateDecision( + allowed=True, + reason="checks passed", + required_checks=["unit-tests"], + human_review_required=False, + metadata={"reviewer": "bot"}, + ) + task = AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=action, + gate_decision=decision, + status="approved", + metadata={"owner": "team-a"}, + ) + + self.assertEqual( + task.to_dict(), + { + "trigger": { + "source": "watcher", + "kind": "strategy_update", + "severity": "high", + "reason": "threshold exceeded", + "subject": "strategy-a", + "metrics": {"score": 0.91}, + "evidence": ["log-1"], + "created_at": 123.4, + }, + "evidence": { + "summary": "summary", + "artifacts": ["artifact-1"], + "metrics": {"latency": 42}, + "risks": ["regression"], + }, + "proposed_action": { + "action": "adjust_threshold", + "lane": "automation", + "target": "strategy-a", + "rationale": "reduce risk", + "requires_human_review": False, + "metadata": {"source": "test"}, + }, + "gate_decision": { + "allowed": True, + "reason": "checks passed", + "required_checks": ["unit-tests"], + "human_review_required": False, + "metadata": {"reviewer": "bot"}, + }, + "status": "approved", + "metadata": {"owner": "team-a"}, + }, + ) + + def test_mutable_defaults_are_not_shared(self) -> None: + first_trigger = TriggerRecord("a", "k", "high", "r", "s") + second_trigger = TriggerRecord("a", "k", "high", "r", "s") + first_trigger.metrics["x"] = 1 + first_trigger.evidence.append("e1") + self.assertEqual(second_trigger.metrics, {}) + self.assertEqual(second_trigger.evidence, []) + + first_bundle = EvidenceBundle("sum") + second_bundle = EvidenceBundle("sum") + first_bundle.artifacts.append("a1") + first_bundle.metrics["m"] = 2 + first_bundle.risks.append("r1") + self.assertEqual(second_bundle.artifacts, []) + self.assertEqual(second_bundle.metrics, {}) + self.assertEqual(second_bundle.risks, []) + + first_action = ProposedAction("act", "lane", "target", "why") + second_action = ProposedAction("act", "lane", "target", "why") + first_action.metadata["x"] = "y" + self.assertEqual(second_action.metadata, {}) + + first_decision = GateDecision(True, "ok") + second_decision = GateDecision(True, "ok") + first_decision.required_checks.append("c1") + first_decision.metadata["x"] = "y" + self.assertEqual(second_decision.required_checks, []) + self.assertEqual(second_decision.metadata, {}) + + first_task = AutomationTask( + trigger=first_trigger, + evidence=first_bundle, + proposed_action=first_action, + gate_decision=first_decision, + ) + second_task = AutomationTask( + trigger=second_trigger, + evidence=second_bundle, + proposed_action=second_action, + gate_decision=second_decision, + ) + first_task.metadata["x"] = 1 + self.assertEqual(second_task.metadata, {}) + + def test_empty_field_validation(self) -> None: + with self.assertRaises(ValueError): + TriggerRecord("a", "k", "", "r", "s") + with self.assertRaises(ValueError): + ProposedAction("", "lane", "target", "why") + with self.assertRaises(ValueError): + AutomationTask( + trigger=TriggerRecord("a", "k", "high", "r", "s"), + evidence=EvidenceBundle("sum"), + proposed_action=ProposedAction("act", "lane", "target", "why"), + gate_decision=GateDecision(True, "ok"), + status="", + ) + + def test_is_actionable(self) -> None: + trigger = TriggerRecord("a", "k", "high", "r", "s") + evidence = EvidenceBundle("sum") + allowed_task = AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=ProposedAction("act", "lane", "target", "why"), + gate_decision=GateDecision(True, "ok"), + ) + blocked_task = AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=ProposedAction("act", "lane", "target", "why"), + gate_decision=GateDecision(False, "blocked"), + ) + + self.assertTrue(allowed_task.is_actionable) + self.assertFalse(blocked_task.is_actionable) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py new file mode 100644 index 00000000..fa0f9f33 --- /dev/null +++ b/tests/test_run_strategy_optimization_watcher.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import tempfile +import unittest + +from scripts.run_strategy_optimization_watcher import parse_bool, resolve_input_path, run_watcher + + +class RunStrategyOptimizationWatcherTest(unittest.TestCase): + def test_dry_run_does_not_create_issue(self) -> None: + calls: list[tuple[str, str, str]] = [] + + result = run_watcher( + { + "repo": "QuantStrategyLab/TestStrategies", + "profile": "live", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + dry_run=True, + create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/1", + find_issue=lambda repo, title: "", + ) + + self.assertEqual(result["findings"], 1) + self.assertTrue(result["issues"][0]["dry_run"]) + self.assertEqual(calls, []) + + def test_non_dry_run_uses_source_repo_override(self) -> None: + calls: list[tuple[str, str, str]] = [] + + result = run_watcher( + { + "repo": "QuantStrategyLab/MetricSource", + "profile": "live", + "current_metrics": {"max_dd": 0.2}, + "baseline_metrics": {"max_dd": 0.1}, + }, + source_repo="QuantStrategyLab/IssueRepo", + dry_run=False, + create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/issue/1", + find_issue=lambda repo, title: "", + ) + + self.assertTrue(result["issues"][0]["created"]) + self.assertEqual(result["issues"][0]["url"], "https://example.test/issue/1") + self.assertEqual(calls[0][0], "QuantStrategyLab/IssueRepo") + + def test_non_dry_run_skips_existing_open_issue(self) -> None: + calls: list[tuple[str, str, str]] = [] + + result = run_watcher( + { + "repo": "QuantStrategyLab/TestStrategies", + "profile": "live", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + dry_run=False, + create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/new", + find_issue=lambda repo, title: "https://example.test/existing", + ) + + self.assertFalse(result["issues"][0]["created"]) + self.assertEqual(result["issues"][0]["existing_url"], "https://example.test/existing") + self.assertEqual(calls, []) + + def test_resolve_input_path_rejects_metrics_path_traversal(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(ValueError): + resolve_input_path(source_root=tmp, metrics_path="../outside.json") + with self.assertRaises(ValueError): + resolve_input_path(source_root=tmp, metrics_path="/tmp/outside.json") + + def test_resolve_input_path_accepts_source_relative_metrics_path(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + resolved = resolve_input_path(source_root=tmp, metrics_path="data/output/strategy_metrics.json") + + self.assertTrue(str(resolved).endswith("data/output/strategy_metrics.json")) + + def test_parse_bool_defaults_safely(self) -> None: + self.assertTrue(parse_bool("true")) + self.assertFalse(parse_bool("false")) + self.assertTrue(parse_bool(None, default=True)) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_strategy_optimization_policy.py b/tests/test_strategy_optimization_policy.py new file mode 100644 index 00000000..c1576f14 --- /dev/null +++ b/tests/test_strategy_optimization_policy.py @@ -0,0 +1,44 @@ +from __future__ import annotations + +import unittest + +from service.strategy_optimization_policy import SEVERITY_HIGH, SEVERITY_MEDIUM, SEVERITY_NONE, evaluate_strategy_metrics + + +class StrategyOptimizationPolicyTest(unittest.TestCase): + def test_detects_multi_metric_degradation_as_high_severity(self) -> None: + result = evaluate_strategy_metrics( + {"sharpe": 0.8, "max_dd": 0.16}, + {"sharpe": 1.0, "max_dd": 0.10}, + ) + + self.assertTrue(result["should_open_issue"]) + self.assertEqual(result["severity"], SEVERITY_HIGH) + self.assertEqual({signal["metric"] for signal in result["signals"]}, {"sharpe", "max_dd"}) + + def test_single_small_degradation_is_medium_severity(self) -> None: + result = evaluate_strategy_metrics({"sharpe": 0.9}, {"sharpe": 1.0}) + + self.assertTrue(result["should_open_issue"]) + self.assertEqual(result["severity"], SEVERITY_MEDIUM) + + def test_baseline_zero_negative_current_is_degradation(self) -> None: + result = evaluate_strategy_metrics({"sharpe": -0.2}, {"sharpe": 0.0}) + + self.assertTrue(result["should_open_issue"]) + self.assertEqual(result["severity"], SEVERITY_MEDIUM) + self.assertEqual(result["signals"][0]["metric"], "sharpe") + + def test_ignores_missing_or_below_threshold_metrics(self) -> None: + result = evaluate_strategy_metrics( + {"sharpe": 0.98, "max_dd": 0.111, "calmar": "not-a-number"}, + {"sharpe": 1.0, "max_dd": 0.10, "calmar": 1.2}, + ) + + self.assertFalse(result["should_open_issue"]) + self.assertEqual(result["severity"], SEVERITY_NONE) + self.assertEqual(result["signals"], []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_strategy_optimization_watcher_workflow.py b/tests/test_strategy_optimization_watcher_workflow.py new file mode 100644 index 00000000..798a3502 --- /dev/null +++ b/tests/test_strategy_optimization_watcher_workflow.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from pathlib import Path +import unittest + + +WORKFLOW_PATH = Path(__file__).resolve().parents[1] / ".github/workflows/strategy_optimization_watcher.yml" + + +class StrategyOptimizationWatcherWorkflowTest(unittest.TestCase): + def test_workflow_is_issue_only_and_dry_run_by_default(self) -> None: + text = WORKFLOW_PATH.read_text(encoding="utf-8") + + self.assertIn("name: Strategy Optimization Watcher", text) + self.assertIn("default: true", text) + self.assertIn("STRATEGY_WATCH_DRY_RUN", text) + self.assertIn("scripts/run_strategy_optimization_watcher.py", text) + self.assertIn("permission-issues: write", text) + self.assertNotIn("pull-request", text.lower()) + self.assertNotIn("auto_merge", text.lower()) + self.assertNotIn("deploy", text.lower()) + + def test_workflow_uses_source_metrics_checkout(self) -> None: + text = WORKFLOW_PATH.read_text(encoding="utf-8") + + self.assertIn("SOURCE_REPO", text) + self.assertIn("METRICS_PATH", text) + self.assertIn("path: source", text) + self.assertIn("STRATEGY_WATCH_SOURCE_ROOT: ${{ github.workspace }}/source", text) + self.assertIn("STRATEGY_WATCH_METRICS_PATH: ${{ env.METRICS_PATH }}", text) + self.assertIn("actions/create-github-app-token", text) + + def test_workflow_fails_closed_for_cross_repo_without_app_token(self) -> None: + text = WORKFLOW_PATH.read_text(encoding="utf-8") + + self.assertIn("^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$", text) + self.assertIn("Verify Source Repository Token", text) + self.assertIn("${SOURCE_REPO}" + '" != "' + "${GITHUB_REPOSITORY}", text) + self.assertIn("Cross-repository strategy watcher requires", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_strategy_watch.py b/tests/test_strategy_watch.py new file mode 100644 index 00000000..5a532c39 --- /dev/null +++ b/tests/test_strategy_watch.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import unittest + +from service.strategy_watch import evaluate_strategy_watch, finding_to_automation_task, issue_for_task + + +class StrategyWatchTest(unittest.TestCase): + def test_degraded_snapshot_becomes_issue_only_task(self) -> None: + findings = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + { + "strategy_profile": "mean_reversion_live", + "plugin": "mean_reversion", + "current_metrics": {"sharpe": 0.7, "max_dd": 0.18}, + "baseline_metrics": {"sharpe": 1.0, "max_dd": 0.1}, + "source": "data/output/strategy_metrics.json", + } + ], + } + ) + + self.assertEqual(len(findings), 1) + task = finding_to_automation_task(findings[0]) + payload = task.to_dict() + self.assertTrue(task.is_actionable) + self.assertEqual(payload["proposed_action"]["action"], "open_issue") + self.assertTrue(payload["proposed_action"]["requires_human_review"]) + self.assertTrue(payload["gate_decision"]["human_review_required"]) + self.assertFalse(payload["gate_decision"]["metadata"]["live_impact_allowed"]) + + def test_healthy_snapshot_creates_no_finding(self) -> None: + findings = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/TestStrategies", + "current_metrics": {"sharpe": 1.01, "max_dd": 0.10}, + "baseline_metrics": {"sharpe": 1.0, "max_dd": 0.1}, + } + ) + + self.assertEqual(findings, []) + + def test_issue_body_states_safety_boundary(self) -> None: + finding = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/TestStrategies", + "profile": "live", + "current_metrics": {"sharpe": 0.8}, + "baseline_metrics": {"sharpe": 1.0}, + } + )[0] + + issue = issue_for_task(finding_to_automation_task(finding)) + + self.assertIn("AI strategy optimization proposal", issue["title"]) + self.assertIn("only opens an issue", issue["body"]) + self.assertIn("does not modify strategy code", issue["body"]) + self.assertIn("sandbox backtest", issue["body"]) + + +if __name__ == "__main__": + unittest.main() From 02c39a0bed3a03b0f8e5d3ea5c571bd4abcd3e6f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 01:33:44 +0800 Subject: [PATCH 2/9] fix: harden strategy watcher workflow gates Co-Authored-By: Codex --- .../strategy_optimization_watcher.yml | 6 ++- scripts/run_strategy_optimization_watcher.py | 38 +++++++++++-------- .../test_run_strategy_optimization_watcher.py | 22 ++++++++++- ..._strategy_optimization_watcher_workflow.py | 4 ++ 4 files changed, 51 insertions(+), 19 deletions(-) diff --git a/.github/workflows/strategy_optimization_watcher.yml b/.github/workflows/strategy_optimization_watcher.yml index ca711cb6..82f72a9e 100644 --- a/.github/workflows/strategy_optimization_watcher.yml +++ b/.github/workflows/strategy_optimization_watcher.yml @@ -40,7 +40,7 @@ jobs: SOURCE_REPO: ${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || github.repository }} SOURCE_REF: ${{ github.event.inputs.source_ref || vars.STRATEGY_WATCH_SOURCE_REF || 'main' }} METRICS_PATH: ${{ github.event.inputs.metrics_path || vars.STRATEGY_WATCH_METRICS_PATH || 'data/output/strategy_metrics.json' }} - STRATEGY_WATCH_DRY_RUN: ${{ github.event.inputs.dry_run || vars.STRATEGY_WATCH_DRY_RUN || 'true' }} + STRATEGY_WATCH_DRY_RUN: ${{ github.event_name == 'workflow_dispatch' && format('{0}', inputs.dry_run) || vars.STRATEGY_WATCH_DRY_RUN || 'true' }} steps: - name: Checkout Bridge uses: actions/checkout@v6.0.3 @@ -69,7 +69,9 @@ jobs: echo "Invalid SOURCE_REPO: ${SOURCE_REPO}. Expected owner/name." >&2 exit 1 fi + owner="${SOURCE_REPO%%/*}" repository="${SOURCE_REPO#*/}" + echo "owner=${owner}" >> "$GITHUB_OUTPUT" echo "repository=${repository}" >> "$GITHUB_OUTPUT" - name: Create GitHub App Token For Source Repository @@ -80,7 +82,7 @@ jobs: with: app-id: ${{ vars.CROSS_REPO_GITHUB_APP_ID }} private-key: ${{ secrets.CROSS_REPO_GITHUB_APP_PRIVATE_KEY }} - owner: ${{ github.repository_owner }} + owner: ${{ steps.source_repo.outputs.owner }} repositories: ${{ steps.source_repo.outputs.repository }} permission-contents: read permission-issues: write diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py index 5e492182..3ab18306 100755 --- a/scripts/run_strategy_optimization_watcher.py +++ b/scripts/run_strategy_optimization_watcher.py @@ -70,22 +70,28 @@ def load_payload(path: str | Path) -> dict[str, Any]: def find_existing_open_issue(repo: str, title: str) -> str: if not REPO_RE.fullmatch(repo): raise ValueError("repository must be in owner/name form") - result = subprocess.run( - ["gh", "issue", "list", "--repo", repo, "--state", "open", "--limit", "1000", "--json", "title,url"], - check=True, - capture_output=True, - text=True, - ) - try: - issues = json.loads(result.stdout or "[]") - except json.JSONDecodeError: - return "" - if not isinstance(issues, list): - return "" - for issue in issues: - if isinstance(issue, dict) and issue.get("title") == title: - return str(issue.get("url") or "") - return "" + page = 1 + while True: + result = subprocess.run( + ["gh", "api", f"/repos/{repo}/issues", "-f", "state=open", "-f", "per_page=100", "-f", f"page={page}"], + check=True, + capture_output=True, + text=True, + ) + try: + issues = json.loads(result.stdout or "[]") + except json.JSONDecodeError: + return "" + if not isinstance(issues, list) or not issues: + return "" + for issue in issues: + if not isinstance(issue, dict) or "pull_request" in issue: + continue + if issue.get("title") == title: + return str(issue.get("html_url") or issue.get("url") or "") + if len(issues) < 100: + return "" + page += 1 def create_github_issue(repo: str, title: str, body: str) -> str: diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py index fa0f9f33..54f55fb6 100644 --- a/tests/test_run_strategy_optimization_watcher.py +++ b/tests/test_run_strategy_optimization_watcher.py @@ -1,9 +1,12 @@ from __future__ import annotations +import json +import subprocess import tempfile import unittest +from unittest.mock import patch -from scripts.run_strategy_optimization_watcher import parse_bool, resolve_input_path, run_watcher +from scripts.run_strategy_optimization_watcher import find_existing_open_issue, parse_bool, resolve_input_path, run_watcher class RunStrategyOptimizationWatcherTest(unittest.TestCase): @@ -78,6 +81,23 @@ def test_resolve_input_path_accepts_source_relative_metrics_path(self) -> None: self.assertTrue(str(resolved).endswith("data/output/strategy_metrics.json")) + def test_find_existing_open_issue_paginates_until_exact_match(self) -> None: + calls: list[list[str]] = [] + first_page = [{"title": f"other-{i}", "html_url": f"https://example.test/{i}"} for i in range(100)] + second_page = [{"title": "target", "html_url": "https://example.test/target"}] + + def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + calls.append(cmd) + page = "2" if "page=2" in cmd else "1" + payload = second_page if page == "2" else first_page + return subprocess.CompletedProcess(cmd, 0, stdout=json.dumps(payload), stderr="") + + with patch("scripts.run_strategy_optimization_watcher.subprocess.run", fake_run): + result = find_existing_open_issue("QuantStrategyLab/TestStrategies", "target") + + self.assertEqual(result, "https://example.test/target") + self.assertEqual(len(calls), 2) + def test_parse_bool_defaults_safely(self) -> None: self.assertTrue(parse_bool("true")) self.assertFalse(parse_bool("false")) diff --git a/tests/test_strategy_optimization_watcher_workflow.py b/tests/test_strategy_optimization_watcher_workflow.py index 798a3502..a920efbc 100644 --- a/tests/test_strategy_optimization_watcher_workflow.py +++ b/tests/test_strategy_optimization_watcher_workflow.py @@ -29,6 +29,8 @@ def test_workflow_uses_source_metrics_checkout(self) -> None: self.assertIn("STRATEGY_WATCH_SOURCE_ROOT: ${{ github.workspace }}/source", text) self.assertIn("STRATEGY_WATCH_METRICS_PATH: ${{ env.METRICS_PATH }}", text) self.assertIn("actions/create-github-app-token", text) + self.assertIn("format('{0}', inputs.dry_run)", text) + self.assertNotIn("github.event.inputs.dry_run ||", text) def test_workflow_fails_closed_for_cross_repo_without_app_token(self) -> None: text = WORKFLOW_PATH.read_text(encoding="utf-8") @@ -37,6 +39,8 @@ def test_workflow_fails_closed_for_cross_repo_without_app_token(self) -> None: self.assertIn("Verify Source Repository Token", text) self.assertIn("${SOURCE_REPO}" + '" != "' + "${GITHUB_REPOSITORY}", text) self.assertIn("Cross-repository strategy watcher requires", text) + self.assertIn("owner=${owner}", text) + self.assertIn("owner: ${{ steps.source_repo.outputs.owner }}", text) if __name__ == "__main__": From 639bbd7df8d60951741f2ae6f9212aa6d0b6d82f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 01:38:23 +0800 Subject: [PATCH 3/9] fix: keep watcher dry-run fail-safe Co-Authored-By: Codex --- .../workflows/strategy_optimization_watcher.yml | 4 ++-- scripts/run_strategy_optimization_watcher.py | 13 +++++++++++-- service/strategy_watch.py | 8 ++++++-- tests/test_run_strategy_optimization_watcher.py | 2 ++ .../test_strategy_optimization_watcher_workflow.py | 2 ++ tests/test_strategy_watch.py | 14 ++++++++++++++ 6 files changed, 37 insertions(+), 6 deletions(-) diff --git a/.github/workflows/strategy_optimization_watcher.yml b/.github/workflows/strategy_optimization_watcher.yml index 82f72a9e..6f69b1d1 100644 --- a/.github/workflows/strategy_optimization_watcher.yml +++ b/.github/workflows/strategy_optimization_watcher.yml @@ -28,7 +28,7 @@ permissions: issues: write concurrency: - group: strategy-optimization-watcher-${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || github.repository }} + group: strategy-optimization-watcher-${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || 'QuantStrategyLab/CryptoLivePoolPipelines' }} cancel-in-progress: false jobs: @@ -37,7 +37,7 @@ jobs: timeout-minutes: 15 env: FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true" - SOURCE_REPO: ${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || github.repository }} + SOURCE_REPO: ${{ github.event.inputs.source_repo || vars.STRATEGY_WATCH_SOURCE_REPO || 'QuantStrategyLab/CryptoLivePoolPipelines' }} SOURCE_REF: ${{ github.event.inputs.source_ref || vars.STRATEGY_WATCH_SOURCE_REF || 'main' }} METRICS_PATH: ${{ github.event.inputs.metrics_path || vars.STRATEGY_WATCH_METRICS_PATH || 'data/output/strategy_metrics.json' }} STRATEGY_WATCH_DRY_RUN: ${{ github.event_name == 'workflow_dispatch' && format('{0}', inputs.dry_run) || vars.STRATEGY_WATCH_DRY_RUN || 'true' }} diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py index 3ab18306..6ee485fb 100755 --- a/scripts/run_strategy_optimization_watcher.py +++ b/scripts/run_strategy_optimization_watcher.py @@ -28,7 +28,11 @@ def parse_bool(value: Any, *, default: bool = False) -> bool: text = str(value).strip().lower() if not text: return default - return text in {"1", "true", "yes", "on"} + if text in {"1", "true", "yes", "on"}: + return True + if text in {"0", "false", "no", "off"}: + return False + raise ValueError("boolean value must be one of true/false/yes/no/on/off/1/0") def resolve_input_path( @@ -162,10 +166,15 @@ def main() -> int: print(json.dumps({"status": "error", "error": "strategy metrics input not found"}, sort_keys=True)) return 2 payload = load_payload(input_path) + try: + dry_run = parse_bool(os.environ.get("STRATEGY_WATCH_DRY_RUN"), default=True) + except ValueError as exc: + print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) + return 2 result = run_watcher( payload, source_repo=os.environ.get("STRATEGY_WATCH_SOURCE_REPO", "").strip(), - dry_run=parse_bool(os.environ.get("STRATEGY_WATCH_DRY_RUN"), default=True), + dry_run=dry_run, ) print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 diff --git a/service/strategy_watch.py b/service/strategy_watch.py index 631c6235..e5f62924 100644 --- a/service/strategy_watch.py +++ b/service/strategy_watch.py @@ -13,6 +13,10 @@ ISSUE_ONLY_ACTION = "open_issue" +def _dict_payload(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, dict) else {} + + @dataclass(frozen=True) class StrategyWatchSnapshot: repo: str @@ -30,8 +34,8 @@ def from_dict(cls, payload: dict[str, Any], *, default_repo: str = "") -> "Strat repo=str(payload.get("repo") or payload.get("repository") or default_repo).strip(), profile=profile, plugin=str(payload.get("plugin") or payload.get("strategy_plugin") or "").strip(), - current_metrics=dict(payload.get("current_metrics") or payload.get("current") or {}), - baseline_metrics=dict(payload.get("baseline_metrics") or payload.get("baseline") or {}), + current_metrics=_dict_payload(payload.get("current_metrics") or payload.get("current")), + baseline_metrics=_dict_payload(payload.get("baseline_metrics") or payload.get("baseline")), source=str(payload.get("source") or "").strip(), generated_at=str(payload.get("generated_at") or "").strip(), ) diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py index 54f55fb6..6f06fb95 100644 --- a/tests/test_run_strategy_optimization_watcher.py +++ b/tests/test_run_strategy_optimization_watcher.py @@ -102,6 +102,8 @@ def test_parse_bool_defaults_safely(self) -> None: self.assertTrue(parse_bool("true")) self.assertFalse(parse_bool("false")) self.assertTrue(parse_bool(None, default=True)) + with self.assertRaises(ValueError): + parse_bool("flase") if __name__ == "__main__": diff --git a/tests/test_strategy_optimization_watcher_workflow.py b/tests/test_strategy_optimization_watcher_workflow.py index a920efbc..e3fd8905 100644 --- a/tests/test_strategy_optimization_watcher_workflow.py +++ b/tests/test_strategy_optimization_watcher_workflow.py @@ -24,6 +24,8 @@ def test_workflow_uses_source_metrics_checkout(self) -> None: text = WORKFLOW_PATH.read_text(encoding="utf-8") self.assertIn("SOURCE_REPO", text) + self.assertIn("QuantStrategyLab/CryptoLivePoolPipelines", text) + self.assertNotIn("vars.STRATEGY_WATCH_SOURCE_REPO || github.repository", text) self.assertIn("METRICS_PATH", text) self.assertIn("path: source", text) self.assertIn("STRATEGY_WATCH_SOURCE_ROOT: ${{ github.workspace }}/source", text) diff --git a/tests/test_strategy_watch.py b/tests/test_strategy_watch.py index 5a532c39..d08915e4 100644 --- a/tests/test_strategy_watch.py +++ b/tests/test_strategy_watch.py @@ -31,6 +31,20 @@ def test_degraded_snapshot_becomes_issue_only_task(self) -> None: self.assertTrue(payload["gate_decision"]["human_review_required"]) self.assertFalse(payload["gate_decision"]["metadata"]["live_impact_allowed"]) + def test_malformed_metrics_snapshot_is_ignored_without_crashing(self) -> None: + findings = evaluate_strategy_watch( + { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + {"profile": "bad", "current_metrics": "oops", "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "live", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + ], + } + ) + + self.assertEqual(len(findings), 1) + self.assertEqual(findings[0].snapshot.profile, "live") + def test_healthy_snapshot_creates_no_finding(self) -> None: findings = evaluate_strategy_watch( { From 682ee86b8fcdb4793e6afb0551f9137242565507 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 01:45:36 +0800 Subject: [PATCH 4/9] fix: restrict strategy watcher source repos Co-Authored-By: Codex --- .../strategy_optimization_watcher.yml | 14 ++++++++ scripts/run_strategy_optimization_watcher.py | 34 ++++++++++++++----- .../test_run_strategy_optimization_watcher.py | 34 +++++++++++++++---- ..._strategy_optimization_watcher_workflow.py | 2 ++ 4 files changed, 69 insertions(+), 15 deletions(-) diff --git a/.github/workflows/strategy_optimization_watcher.yml b/.github/workflows/strategy_optimization_watcher.yml index 6f69b1d1..48e3d9e7 100644 --- a/.github/workflows/strategy_optimization_watcher.yml +++ b/.github/workflows/strategy_optimization_watcher.yml @@ -41,6 +41,7 @@ jobs: SOURCE_REF: ${{ github.event.inputs.source_ref || vars.STRATEGY_WATCH_SOURCE_REF || 'main' }} METRICS_PATH: ${{ github.event.inputs.metrics_path || vars.STRATEGY_WATCH_METRICS_PATH || 'data/output/strategy_metrics.json' }} STRATEGY_WATCH_DRY_RUN: ${{ github.event_name == 'workflow_dispatch' && format('{0}', inputs.dry_run) || vars.STRATEGY_WATCH_DRY_RUN || 'true' }} + ALLOWED_SOURCE_REPOS: ${{ vars.STRATEGY_WATCH_ALLOWED_SOURCE_REPOS || 'QuantStrategyLab/CryptoLivePoolPipelines' }} steps: - name: Checkout Bridge uses: actions/checkout@v6.0.3 @@ -69,6 +70,19 @@ jobs: echo "Invalid SOURCE_REPO: ${SOURCE_REPO}. Expected owner/name." >&2 exit 1 fi + allowed_match=false + IFS=',' read -ra allowed_repos <<< "${ALLOWED_SOURCE_REPOS}" + for allowed_repo in "${allowed_repos[@]}"; do + allowed_repo="${allowed_repo//[[:space:]]/}" + if [ "${allowed_repo}" = "${SOURCE_REPO}" ]; then + allowed_match=true + break + fi + done + if [ "${allowed_match}" != "true" ]; then + echo "SOURCE_REPO is not allowed for strategy watcher: ${SOURCE_REPO}" >&2 + exit 1 + fi owner="${SOURCE_REPO%%/*}" repository="${SOURCE_REPO#*/}" echo "owner=${owner}" >> "$GITHUB_OUTPUT" diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py index 6ee485fb..11a84cec 100755 --- a/scripts/run_strategy_optimization_watcher.py +++ b/scripts/run_strategy_optimization_watcher.py @@ -71,10 +71,11 @@ def load_payload(path: str | Path) -> dict[str, Any]: return payload -def find_existing_open_issue(repo: str, title: str) -> str: +def list_open_issue_urls(repo: str) -> dict[str, str]: if not REPO_RE.fullmatch(repo): raise ValueError("repository must be in owner/name form") page = 1 + open_issues: dict[str, str] = {} while True: result = subprocess.run( ["gh", "api", f"/repos/{repo}/issues", "-f", "state=open", "-f", "per_page=100", "-f", f"page={page}"], @@ -85,19 +86,24 @@ def find_existing_open_issue(repo: str, title: str) -> str: try: issues = json.loads(result.stdout or "[]") except json.JSONDecodeError: - return "" + return open_issues if not isinstance(issues, list) or not issues: - return "" + return open_issues for issue in issues: if not isinstance(issue, dict) or "pull_request" in issue: continue - if issue.get("title") == title: - return str(issue.get("html_url") or issue.get("url") or "") + title = str(issue.get("title") or "") + if title and title not in open_issues: + open_issues[title] = str(issue.get("html_url") or issue.get("url") or "") if len(issues) < 100: - return "" + return open_issues page += 1 +def find_existing_open_issue(repo: str, title: str) -> str: + return list_open_issue_urls(repo).get(title, "") + + def create_github_issue(repo: str, title: str, body: str) -> str: if not REPO_RE.fullmatch(repo): raise ValueError("repository must be in owner/name form") @@ -116,10 +122,11 @@ def run_watcher( source_repo: str = "", dry_run: bool = True, create_issue: Callable[[str, str, str], str] = create_github_issue, - find_issue: Callable[[str, str], str] = find_existing_open_issue, + list_issues: Callable[[str], dict[str, str]] = list_open_issue_urls, ) -> dict[str, Any]: findings = evaluate_strategy_watch(payload) issues: list[dict[str, Any]] = [] + open_issue_cache: dict[str, dict[str, str]] = {} for finding in findings: task = finding_to_automation_task(finding) issue = issue_for_task(task) @@ -133,7 +140,9 @@ def run_watcher( if dry_run: issue_result["dry_run"] = True else: - existing_url = find_issue(repo, issue["title"]) + if repo not in open_issue_cache: + open_issue_cache[repo] = list_issues(repo) + existing_url = open_issue_cache[repo].get(issue["title"], "") if existing_url: issue_result["existing_url"] = existing_url issue_result["skipped_reason"] = "open issue already exists" @@ -165,7 +174,14 @@ def main() -> int: if not input_path.exists(): print(json.dumps({"status": "error", "error": "strategy metrics input not found"}, sort_keys=True)) return 2 - payload = load_payload(input_path) + if not input_path.is_file(): + print(json.dumps({"status": "error", "error": "strategy metrics input is not a file"}, sort_keys=True)) + return 2 + try: + payload = load_payload(input_path) + except (OSError, json.JSONDecodeError, ValueError) as exc: + print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) + return 2 try: dry_run = parse_bool(os.environ.get("STRATEGY_WATCH_DRY_RUN"), default=True) except ValueError as exc: diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py index 6f06fb95..819b3b7c 100644 --- a/tests/test_run_strategy_optimization_watcher.py +++ b/tests/test_run_strategy_optimization_watcher.py @@ -6,7 +6,7 @@ import unittest from unittest.mock import patch -from scripts.run_strategy_optimization_watcher import find_existing_open_issue, parse_bool, resolve_input_path, run_watcher +from scripts.run_strategy_optimization_watcher import list_open_issue_urls, parse_bool, resolve_input_path, run_watcher class RunStrategyOptimizationWatcherTest(unittest.TestCase): @@ -22,7 +22,7 @@ def test_dry_run_does_not_create_issue(self) -> None: }, dry_run=True, create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/1", - find_issue=lambda repo, title: "", + list_issues=lambda repo: {}, ) self.assertEqual(result["findings"], 1) @@ -42,7 +42,7 @@ def test_non_dry_run_uses_source_repo_override(self) -> None: source_repo="QuantStrategyLab/IssueRepo", dry_run=False, create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/issue/1", - find_issue=lambda repo, title: "", + list_issues=lambda repo: {}, ) self.assertTrue(result["issues"][0]["created"]) @@ -61,7 +61,7 @@ def test_non_dry_run_skips_existing_open_issue(self) -> None: }, dry_run=False, create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/new", - find_issue=lambda repo, title: "https://example.test/existing", + list_issues=lambda repo: {"AI strategy optimization proposal: QuantStrategyLab/TestStrategies:live": "https://example.test/existing"}, ) self.assertFalse(result["issues"][0]["created"]) @@ -93,11 +93,33 @@ def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[st return subprocess.CompletedProcess(cmd, 0, stdout=json.dumps(payload), stderr="") with patch("scripts.run_strategy_optimization_watcher.subprocess.run", fake_run): - result = find_existing_open_issue("QuantStrategyLab/TestStrategies", "target") + issues = list_open_issue_urls("QuantStrategyLab/TestStrategies") - self.assertEqual(result, "https://example.test/target") + self.assertEqual(issues["target"], "https://example.test/target") self.assertEqual(len(calls), 2) + def test_run_watcher_caches_open_issues_per_repo(self) -> None: + list_calls: list[str] = [] + create_calls: list[tuple[str, str, str]] = [] + payload = { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + {"profile": "a", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "b", "current_metrics": {"sharpe": 0.4}, "baseline_metrics": {"sharpe": 1.0}}, + ], + } + + result = run_watcher( + payload, + dry_run=False, + create_issue=lambda repo, title, body: create_calls.append((repo, title, body)) or "https://example.test/new", + list_issues=lambda repo: list_calls.append(repo) or {}, + ) + + self.assertEqual(result["findings"], 2) + self.assertEqual(list_calls, ["QuantStrategyLab/TestStrategies"]) + self.assertEqual(len(create_calls), 2) + def test_parse_bool_defaults_safely(self) -> None: self.assertTrue(parse_bool("true")) self.assertFalse(parse_bool("false")) diff --git a/tests/test_strategy_optimization_watcher_workflow.py b/tests/test_strategy_optimization_watcher_workflow.py index e3fd8905..1bd70faf 100644 --- a/tests/test_strategy_optimization_watcher_workflow.py +++ b/tests/test_strategy_optimization_watcher_workflow.py @@ -25,6 +25,7 @@ def test_workflow_uses_source_metrics_checkout(self) -> None: self.assertIn("SOURCE_REPO", text) self.assertIn("QuantStrategyLab/CryptoLivePoolPipelines", text) + self.assertIn("STRATEGY_WATCH_ALLOWED_SOURCE_REPOS", text) self.assertNotIn("vars.STRATEGY_WATCH_SOURCE_REPO || github.repository", text) self.assertIn("METRICS_PATH", text) self.assertIn("path: source", text) @@ -41,6 +42,7 @@ def test_workflow_fails_closed_for_cross_repo_without_app_token(self) -> None: self.assertIn("Verify Source Repository Token", text) self.assertIn("${SOURCE_REPO}" + '" != "' + "${GITHUB_REPOSITORY}", text) self.assertIn("Cross-repository strategy watcher requires", text) + self.assertIn("SOURCE_REPO is not allowed", text) self.assertIn("owner=${owner}", text) self.assertIn("owner: ${{ steps.source_repo.outputs.owner }}", text) From a29d2b30edd440f080e4b03e1ec9ff7ebb934045 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 01:51:38 +0800 Subject: [PATCH 5/9] fix: redact watcher diagnostics Co-Authored-By: Codex --- scripts/run_strategy_optimization_watcher.py | 31 +++++++++++++++++-- .../test_run_strategy_optimization_watcher.py | 24 ++++++++++++++ 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py index 11a84cec..77f3e944 100755 --- a/scripts/run_strategy_optimization_watcher.py +++ b/scripts/run_strategy_optimization_watcher.py @@ -78,7 +78,7 @@ def list_open_issue_urls(repo: str) -> dict[str, str]: open_issues: dict[str, str] = {} while True: result = subprocess.run( - ["gh", "api", f"/repos/{repo}/issues", "-f", "state=open", "-f", "per_page=100", "-f", f"page={page}"], + ["gh", "api", "--method", "GET", f"/repos/{repo}/issues", "-f", "state=open", "-f", "per_page=100", "-f", f"page={page}"], check=True, capture_output=True, text=True, @@ -104,6 +104,32 @@ def find_existing_open_issue(repo: str, title: str) -> str: return list_open_issue_urls(repo).get(title, "") +def task_public_summary(task: Any) -> dict[str, Any]: + payload = task.to_dict() + trigger = payload.get("trigger") if isinstance(payload.get("trigger"), dict) else {} + proposed_action = payload.get("proposed_action") if isinstance(payload.get("proposed_action"), dict) else {} + gate_decision = payload.get("gate_decision") if isinstance(payload.get("gate_decision"), dict) else {} + return { + "trigger": { + "source": trigger.get("source", ""), + "kind": trigger.get("kind", ""), + "severity": trigger.get("severity", ""), + "subject": trigger.get("subject", ""), + }, + "proposed_action": { + "action": proposed_action.get("action", ""), + "lane": proposed_action.get("lane", ""), + "target": proposed_action.get("target", ""), + "requires_human_review": proposed_action.get("requires_human_review", True), + }, + "gate_decision": { + "allowed": gate_decision.get("allowed", False), + "human_review_required": gate_decision.get("human_review_required", True), + }, + "status": payload.get("status", ""), + } + + def create_github_issue(repo: str, title: str, body: str) -> str: if not REPO_RE.fullmatch(repo): raise ValueError("repository must be in owner/name form") @@ -134,7 +160,7 @@ def run_watcher( issue_result: dict[str, Any] = { "repo": repo, "title": issue["title"], - "task": task.to_dict(), + "task": task_public_summary(task), "created": False, } if dry_run: @@ -148,6 +174,7 @@ def run_watcher( issue_result["skipped_reason"] = "open issue already exists" else: issue_result["url"] = create_issue(repo, issue["title"], issue["body"]) + open_issue_cache[repo][issue["title"]] = str(issue_result["url"]) issue_result["created"] = True issues.append(issue_result) return { diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py index 819b3b7c..c3fb0809 100644 --- a/tests/test_run_strategy_optimization_watcher.py +++ b/tests/test_run_strategy_optimization_watcher.py @@ -27,6 +27,7 @@ def test_dry_run_does_not_create_issue(self) -> None: self.assertEqual(result["findings"], 1) self.assertTrue(result["issues"][0]["dry_run"]) + self.assertNotIn("metrics", result["issues"][0]["task"]["trigger"]) self.assertEqual(calls, []) def test_non_dry_run_uses_source_repo_override(self) -> None: @@ -97,6 +98,29 @@ def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[st self.assertEqual(issues["target"], "https://example.test/target") self.assertEqual(len(calls), 2) + self.assertIn("--method", calls[0]) + self.assertIn("GET", calls[0]) + + def test_run_watcher_updates_cache_after_create(self) -> None: + create_calls: list[tuple[str, str, str]] = [] + payload = { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + {"profile": "same", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "same", "current_metrics": {"sharpe": 0.4}, "baseline_metrics": {"sharpe": 1.0}}, + ], + } + + result = run_watcher( + payload, + dry_run=False, + create_issue=lambda repo, title, body: create_calls.append((repo, title, body)) or "https://example.test/new", + list_issues=lambda repo: {}, + ) + + self.assertEqual(result["findings"], 2) + self.assertEqual(len(create_calls), 1) + self.assertEqual(result["issues"][1]["existing_url"], "https://example.test/new") def test_run_watcher_caches_open_issues_per_repo(self) -> None: list_calls: list[str] = [] From 82283ec05554477a8126c6e14671bdfa7bc973e3 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 01:56:31 +0800 Subject: [PATCH 6/9] fix: gate strategy watcher refs Co-Authored-By: Codex --- .../strategy_optimization_watcher.yml | 14 ++++++ scripts/run_strategy_optimization_watcher.py | 49 ++++++++++++++++--- .../test_run_strategy_optimization_watcher.py | 35 ++++++++++++- ..._strategy_optimization_watcher_workflow.py | 2 + 4 files changed, 91 insertions(+), 9 deletions(-) diff --git a/.github/workflows/strategy_optimization_watcher.yml b/.github/workflows/strategy_optimization_watcher.yml index 48e3d9e7..4909886a 100644 --- a/.github/workflows/strategy_optimization_watcher.yml +++ b/.github/workflows/strategy_optimization_watcher.yml @@ -42,6 +42,7 @@ jobs: METRICS_PATH: ${{ github.event.inputs.metrics_path || vars.STRATEGY_WATCH_METRICS_PATH || 'data/output/strategy_metrics.json' }} STRATEGY_WATCH_DRY_RUN: ${{ github.event_name == 'workflow_dispatch' && format('{0}', inputs.dry_run) || vars.STRATEGY_WATCH_DRY_RUN || 'true' }} ALLOWED_SOURCE_REPOS: ${{ vars.STRATEGY_WATCH_ALLOWED_SOURCE_REPOS || 'QuantStrategyLab/CryptoLivePoolPipelines' }} + ALLOWED_SOURCE_REFS: ${{ vars.STRATEGY_WATCH_ALLOWED_SOURCE_REFS || 'main' }} steps: - name: Checkout Bridge uses: actions/checkout@v6.0.3 @@ -83,6 +84,19 @@ jobs: echo "SOURCE_REPO is not allowed for strategy watcher: ${SOURCE_REPO}" >&2 exit 1 fi + ref_allowed=false + IFS=',' read -ra allowed_refs <<< "${ALLOWED_SOURCE_REFS}" + for allowed_ref in "${allowed_refs[@]}"; do + allowed_ref="${allowed_ref//[[:space:]]/}" + if [ "${allowed_ref}" = "${SOURCE_REF}" ]; then + ref_allowed=true + break + fi + done + if [ "${ref_allowed}" != "true" ]; then + echo "SOURCE_REF is not allowed for strategy watcher: ${SOURCE_REF}" >&2 + exit 1 + fi owner="${SOURCE_REPO%%/*}" repository="${SOURCE_REPO#*/}" echo "owner=${owner}" >> "$GITHUB_OUTPUT" diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py index 77f3e944..93b4b963 100755 --- a/scripts/run_strategy_optimization_watcher.py +++ b/scripts/run_strategy_optimization_watcher.py @@ -3,6 +3,7 @@ from __future__ import annotations +import copy import json import os from pathlib import Path, PurePosixPath @@ -85,8 +86,8 @@ def list_open_issue_urls(repo: str) -> dict[str, str]: ) try: issues = json.loads(result.stdout or "[]") - except json.JSONDecodeError: - return open_issues + except json.JSONDecodeError as exc: + raise RuntimeError("failed to parse open issue list") from exc if not isinstance(issues, list) or not issues: return open_issues for issue in issues: @@ -142,6 +143,33 @@ def create_github_issue(repo: str, title: str, body: str) -> str: return result.stdout.strip() +def _payload_for_source_repo(payload: dict[str, Any], source_repo: str) -> dict[str, Any]: + if not source_repo: + return payload + normalized = copy.deepcopy(payload) + for key in ("repo", "repository"): + embedded = str(normalized.get(key) or "").strip() + if embedded and embedded != source_repo: + raise ValueError("metrics payload repository does not match validated source repository") + normalized["repo"] = source_repo + raw_snapshots = normalized.get("snapshots") + if isinstance(raw_snapshots, list): + clean_snapshots: list[Any] = [] + for item in raw_snapshots: + if not isinstance(item, dict): + clean_snapshots.append(item) + continue + snapshot = dict(item) + for key in ("repo", "repository"): + embedded = str(snapshot.get(key) or "").strip() + if embedded and embedded != source_repo: + raise ValueError("snapshot repository does not match validated source repository") + snapshot["repo"] = source_repo + clean_snapshots.append(snapshot) + normalized["snapshots"] = clean_snapshots + return normalized + + def run_watcher( payload: dict[str, Any], *, @@ -150,7 +178,8 @@ def run_watcher( create_issue: Callable[[str, str, str], str] = create_github_issue, list_issues: Callable[[str], dict[str, str]] = list_open_issue_urls, ) -> dict[str, Any]: - findings = evaluate_strategy_watch(payload) + watch_payload = _payload_for_source_repo(payload, source_repo) + findings = evaluate_strategy_watch(watch_payload) issues: list[dict[str, Any]] = [] open_issue_cache: dict[str, dict[str, str]] = {} for finding in findings: @@ -214,11 +243,15 @@ def main() -> int: except ValueError as exc: print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) return 2 - result = run_watcher( - payload, - source_repo=os.environ.get("STRATEGY_WATCH_SOURCE_REPO", "").strip(), - dry_run=dry_run, - ) + try: + result = run_watcher( + payload, + source_repo=os.environ.get("STRATEGY_WATCH_SOURCE_REPO", "").strip(), + dry_run=dry_run, + ) + except (ValueError, RuntimeError, subprocess.CalledProcessError) as exc: + print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) + return 2 print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py index c3fb0809..342a1ec6 100644 --- a/tests/test_run_strategy_optimization_watcher.py +++ b/tests/test_run_strategy_optimization_watcher.py @@ -35,7 +35,7 @@ def test_non_dry_run_uses_source_repo_override(self) -> None: result = run_watcher( { - "repo": "QuantStrategyLab/MetricSource", + "repo": "QuantStrategyLab/IssueRepo", "profile": "live", "current_metrics": {"max_dd": 0.2}, "baseline_metrics": {"max_dd": 0.1}, @@ -101,6 +101,31 @@ def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[st self.assertIn("--method", calls[0]) self.assertIn("GET", calls[0]) + def test_run_watcher_rejects_mismatched_source_repo_payload(self) -> None: + with self.assertRaises(ValueError): + run_watcher( + { + "repo": "QuantStrategyLab/Other", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + source_repo="QuantStrategyLab/TestStrategies", + ) + + def test_run_watcher_uses_validated_source_repo_in_task_summary(self) -> None: + result = run_watcher( + { + "profile": "live", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + source_repo="QuantStrategyLab/TestStrategies", + dry_run=True, + ) + + self.assertIn("QuantStrategyLab/TestStrategies:live", result["issues"][0]["title"]) + self.assertEqual(result["issues"][0]["task"]["proposed_action"]["target"], "QuantStrategyLab/TestStrategies") + def test_run_watcher_updates_cache_after_create(self) -> None: create_calls: list[tuple[str, str, str]] = [] payload = { @@ -144,6 +169,14 @@ def test_run_watcher_caches_open_issues_per_repo(self) -> None: self.assertEqual(list_calls, ["QuantStrategyLab/TestStrategies"]) self.assertEqual(len(create_calls), 2) + def test_list_open_issue_urls_fails_closed_on_bad_json(self) -> None: + def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess(cmd, 0, stdout="not-json", stderr="") + + with patch("scripts.run_strategy_optimization_watcher.subprocess.run", fake_run): + with self.assertRaises(RuntimeError): + list_open_issue_urls("QuantStrategyLab/TestStrategies") + def test_parse_bool_defaults_safely(self) -> None: self.assertTrue(parse_bool("true")) self.assertFalse(parse_bool("false")) diff --git a/tests/test_strategy_optimization_watcher_workflow.py b/tests/test_strategy_optimization_watcher_workflow.py index 1bd70faf..8d706120 100644 --- a/tests/test_strategy_optimization_watcher_workflow.py +++ b/tests/test_strategy_optimization_watcher_workflow.py @@ -26,6 +26,8 @@ def test_workflow_uses_source_metrics_checkout(self) -> None: self.assertIn("SOURCE_REPO", text) self.assertIn("QuantStrategyLab/CryptoLivePoolPipelines", text) self.assertIn("STRATEGY_WATCH_ALLOWED_SOURCE_REPOS", text) + self.assertIn("STRATEGY_WATCH_ALLOWED_SOURCE_REFS", text) + self.assertIn("SOURCE_REF is not allowed", text) self.assertNotIn("vars.STRATEGY_WATCH_SOURCE_REPO || github.repository", text) self.assertIn("METRICS_PATH", text) self.assertIn("path: source", text) From dab18769347e40bfbae4dd01a8e0a91a726ff15c Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 02:03:50 +0800 Subject: [PATCH 7/9] fix: key strategy watcher issue events Co-Authored-By: Codex --- scripts/run_strategy_optimization_watcher.py | 27 ++++++++----- service/strategy_watch.py | 28 +++++++++++-- .../test_run_strategy_optimization_watcher.py | 40 +++++++++++++++---- tests/test_strategy_watch.py | 2 + 4 files changed, 75 insertions(+), 22 deletions(-) diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py index 93b4b963..9e0a11a3 100755 --- a/scripts/run_strategy_optimization_watcher.py +++ b/scripts/run_strategy_optimization_watcher.py @@ -195,22 +195,27 @@ def run_watcher( if dry_run: issue_result["dry_run"] = True else: - if repo not in open_issue_cache: - open_issue_cache[repo] = list_issues(repo) - existing_url = open_issue_cache[repo].get(issue["title"], "") - if existing_url: - issue_result["existing_url"] = existing_url - issue_result["skipped_reason"] = "open issue already exists" - else: - issue_result["url"] = create_issue(repo, issue["title"], issue["body"]) - open_issue_cache[repo][issue["title"]] = str(issue_result["url"]) - issue_result["created"] = True + try: + if repo not in open_issue_cache: + open_issue_cache[repo] = list_issues(repo) + existing_url = open_issue_cache[repo].get(issue["title"], "") + if existing_url: + issue_result["existing_url"] = existing_url + issue_result["skipped_reason"] = "open issue already exists" + else: + issue_result["url"] = create_issue(repo, issue["title"], issue["body"]) + open_issue_cache[repo][issue["title"]] = str(issue_result["url"]) + issue_result["created"] = True + except (ValueError, RuntimeError, subprocess.CalledProcessError) as exc: + issue_result["error"] = str(exc) issues.append(issue_result) + errors = sum(1 for issue in issues if issue.get("error")) return { - "status": "ok", + "status": "partial_error" if errors else "ok", "dry_run": dry_run, "findings": len(findings), "issues": issues, + "errors": errors, } diff --git a/service/strategy_watch.py b/service/strategy_watch.py index e5f62924..8c36cd28 100644 --- a/service/strategy_watch.py +++ b/service/strategy_watch.py @@ -3,6 +3,8 @@ from __future__ import annotations from dataclasses import dataclass, field +import hashlib +import json from typing import Any from service.automation_contracts import AutomationTask, EvidenceBundle, GateDecision, ProposedAction, TriggerRecord @@ -105,9 +107,20 @@ def evaluate_strategy_watch(payload: dict[str, Any]) -> list[StrategyWatchFindin return findings +def finding_event_key(finding: StrategyWatchFinding) -> str: + payload = { + "snapshot": finding.snapshot.to_dict(), + "severity": finding.severity, + "signals": finding.signals, + } + raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + return hashlib.sha256(raw).hexdigest()[:12] + + def finding_to_automation_task(finding: StrategyWatchFinding) -> AutomationTask: """Convert a deterministic finding into an issue-only automation task.""" lane = str(finding.registry_context.get("automation_lane") or LANE_RESEARCH_BACKLOG) + event_key = finding_event_key(finding) signal_reasons = [str(signal.get("reason") or signal.get("metric") or "metric degraded") for signal in finding.signals] trigger = TriggerRecord( source="strategy_optimization_watcher", @@ -136,7 +149,7 @@ def finding_to_automation_task(finding: StrategyWatchFinding) -> AutomationTask: target=finding.snapshot.repo, rationale="Open a research optimization issue for AI diagnosis and sandbox experiment planning.", requires_human_review=True, - metadata={"profile": finding.snapshot.profile, "plugin": finding.snapshot.plugin}, + metadata={"profile": finding.snapshot.profile, "plugin": finding.snapshot.plugin, "event_key": event_key}, ) gate = GateDecision( allowed=True, @@ -149,7 +162,13 @@ def finding_to_automation_task(finding: StrategyWatchFinding) -> AutomationTask: human_review_required=True, metadata={"issue_only": True, "live_impact_allowed": False}, ) - return AutomationTask(trigger=trigger, evidence=evidence, proposed_action=proposed, gate_decision=gate) + return AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=proposed, + gate_decision=gate, + metadata={"event_key": event_key}, + ) def issue_for_task(task: AutomationTask) -> dict[str, str]: @@ -158,7 +177,9 @@ def issue_for_task(task: AutomationTask) -> dict[str, str]: trigger = payload["trigger"] evidence = payload["evidence"] action = payload["proposed_action"] - title = f"AI strategy optimization proposal: {trigger.get('subject') or action.get('target') or 'strategy profile'}" + event_key = str(payload.get("metadata", {}).get("event_key") or "") + marker = f" [{event_key}]" if event_key else "" + title = f"AI strategy optimization proposal: {trigger.get('subject') or action.get('target') or 'strategy profile'}{marker}" signals = "\n".join(f"- {item}" for item in trigger.get("evidence", [])) or "- Strategy metrics degraded." checks = "\n".join(f"- [ ] {item}" for item in payload["gate_decision"].get("required_checks", [])) risks = "\n".join(f"- {item}" for item in evidence.get("risks", [])) @@ -170,6 +191,7 @@ def issue_for_task(task: AutomationTask) -> dict[str, str]: "## Trigger", f"- Severity: `{trigger.get('severity')}`", f"- Subject: `{trigger.get('subject')}`", + f"- Event key: `{event_key}`", "", "## Signals", signals, diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py index 342a1ec6..c7b6789a 100644 --- a/tests/test_run_strategy_optimization_watcher.py +++ b/tests/test_run_strategy_optimization_watcher.py @@ -26,6 +26,7 @@ def test_dry_run_does_not_create_issue(self) -> None: ) self.assertEqual(result["findings"], 1) + self.assertEqual(result["errors"], 0) self.assertTrue(result["issues"][0]["dry_run"]) self.assertNotIn("metrics", result["issues"][0]["task"]["trigger"]) self.assertEqual(calls, []) @@ -52,17 +53,19 @@ def test_non_dry_run_uses_source_repo_override(self) -> None: def test_non_dry_run_skips_existing_open_issue(self) -> None: calls: list[tuple[str, str, str]] = [] + payload = { + "repo": "QuantStrategyLab/TestStrategies", + "profile": "live", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + } + title = run_watcher(payload, dry_run=True)["issues"][0]["title"] result = run_watcher( - { - "repo": "QuantStrategyLab/TestStrategies", - "profile": "live", - "current_metrics": {"sharpe": 0.5}, - "baseline_metrics": {"sharpe": 1.0}, - }, + payload, dry_run=False, create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/new", - list_issues=lambda repo: {"AI strategy optimization proposal: QuantStrategyLab/TestStrategies:live": "https://example.test/existing"}, + list_issues=lambda repo: {title: "https://example.test/existing"}, ) self.assertFalse(result["issues"][0]["created"]) @@ -124,15 +127,36 @@ def test_run_watcher_uses_validated_source_repo_in_task_summary(self) -> None: ) self.assertIn("QuantStrategyLab/TestStrategies:live", result["issues"][0]["title"]) + self.assertRegex(result["issues"][0]["title"], r"\[[a-f0-9]{12}\]$") self.assertEqual(result["issues"][0]["task"]["proposed_action"]["target"], "QuantStrategyLab/TestStrategies") + def test_run_watcher_records_issue_errors_per_finding(self) -> None: + payload = { + "repo": "QuantStrategyLab/TestStrategies", + "snapshots": [ + {"profile": "a", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "b", "current_metrics": {"sharpe": 0.4}, "baseline_metrics": {"sharpe": 1.0}}, + ], + } + + result = run_watcher( + payload, + dry_run=False, + create_issue=lambda repo, title, body: (_ for _ in ()).throw(RuntimeError("boom")) if ":a " in title else "https://example.test/new", + list_issues=lambda repo: {}, + ) + + self.assertEqual(result["status"], "partial_error") + self.assertEqual(result["errors"], 1) + self.assertEqual(len(result["issues"]), 2) + def test_run_watcher_updates_cache_after_create(self) -> None: create_calls: list[tuple[str, str, str]] = [] payload = { "repo": "QuantStrategyLab/TestStrategies", "snapshots": [ {"profile": "same", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, - {"profile": "same", "current_metrics": {"sharpe": 0.4}, "baseline_metrics": {"sharpe": 1.0}}, + {"profile": "same", "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}}, ], } diff --git a/tests/test_strategy_watch.py b/tests/test_strategy_watch.py index d08915e4..38372ca0 100644 --- a/tests/test_strategy_watch.py +++ b/tests/test_strategy_watch.py @@ -69,6 +69,8 @@ def test_issue_body_states_safety_boundary(self) -> None: issue = issue_for_task(finding_to_automation_task(finding)) self.assertIn("AI strategy optimization proposal", issue["title"]) + self.assertRegex(issue["title"], r"\[[a-f0-9]{12}\]$") + self.assertIn("Event key", issue["body"]) self.assertIn("only opens an issue", issue["body"]) self.assertIn("does not modify strategy code", issue["body"]) self.assertIn("sandbox backtest", issue["body"]) From 61775f7b7708d7c948bf6b696cdf74e76adb8925 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 02:09:46 +0800 Subject: [PATCH 8/9] fix: comment existing watcher issues Co-Authored-By: Codex --- scripts/run_strategy_optimization_watcher.py | 19 +++++++++++++++++-- service/strategy_watch.py | 3 +-- .../test_run_strategy_optimization_watcher.py | 11 +++++++++-- tests/test_strategy_watch.py | 2 +- 4 files changed, 28 insertions(+), 7 deletions(-) diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py index 9e0a11a3..3e17368d 100755 --- a/scripts/run_strategy_optimization_watcher.py +++ b/scripts/run_strategy_optimization_watcher.py @@ -131,6 +131,18 @@ def task_public_summary(task: Any) -> dict[str, Any]: } +def comment_github_issue(repo: str, issue_url: str, body: str) -> str: + if not REPO_RE.fullmatch(repo): + raise ValueError("repository must be in owner/name form") + result = subprocess.run( + ["gh", "issue", "comment", issue_url, "--repo", repo, "--body", body], + check=True, + capture_output=True, + text=True, + ) + return result.stdout.strip() + + def create_github_issue(repo: str, title: str, body: str) -> str: if not REPO_RE.fullmatch(repo): raise ValueError("repository must be in owner/name form") @@ -176,6 +188,7 @@ def run_watcher( source_repo: str = "", dry_run: bool = True, create_issue: Callable[[str, str, str], str] = create_github_issue, + comment_issue: Callable[[str, str, str], str] = comment_github_issue, list_issues: Callable[[str], dict[str, str]] = list_open_issue_urls, ) -> dict[str, Any]: watch_payload = _payload_for_source_repo(payload, source_repo) @@ -201,7 +214,9 @@ def run_watcher( existing_url = open_issue_cache[repo].get(issue["title"], "") if existing_url: issue_result["existing_url"] = existing_url - issue_result["skipped_reason"] = "open issue already exists" + issue_result["comment_url"] = comment_issue(repo, existing_url, issue["body"]) + issue_result["commented"] = True + issue_result["skipped_reason"] = "open issue already exists; appended watcher update" else: issue_result["url"] = create_issue(repo, issue["title"], issue["body"]) open_issue_cache[repo][issue["title"]] = str(issue_result["url"]) @@ -258,7 +273,7 @@ def main() -> int: print(json.dumps({"status": "error", "error": str(exc)}, sort_keys=True)) return 2 print(json.dumps(result, ensure_ascii=False, sort_keys=True)) - return 0 + return 1 if int(result.get("errors", 0)) > 0 or result.get("status") != "ok" else 0 if __name__ == "__main__": diff --git a/service/strategy_watch.py b/service/strategy_watch.py index 8c36cd28..4b742513 100644 --- a/service/strategy_watch.py +++ b/service/strategy_watch.py @@ -178,8 +178,7 @@ def issue_for_task(task: AutomationTask) -> dict[str, str]: evidence = payload["evidence"] action = payload["proposed_action"] event_key = str(payload.get("metadata", {}).get("event_key") or "") - marker = f" [{event_key}]" if event_key else "" - title = f"AI strategy optimization proposal: {trigger.get('subject') or action.get('target') or 'strategy profile'}{marker}" + title = f"AI strategy optimization proposal: {trigger.get('subject') or action.get('target') or 'strategy profile'}" signals = "\n".join(f"- {item}" for item in trigger.get("evidence", [])) or "- Strategy metrics degraded." checks = "\n".join(f"- [ ] {item}" for item in payload["gate_decision"].get("required_checks", [])) risks = "\n".join(f"- {item}" for item in evidence.get("risks", [])) diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py index c7b6789a..c8885a8b 100644 --- a/tests/test_run_strategy_optimization_watcher.py +++ b/tests/test_run_strategy_optimization_watcher.py @@ -22,6 +22,7 @@ def test_dry_run_does_not_create_issue(self) -> None: }, dry_run=True, create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/1", + comment_issue=lambda repo, url, body: "https://example.test/comment", list_issues=lambda repo: {}, ) @@ -44,6 +45,7 @@ def test_non_dry_run_uses_source_repo_override(self) -> None: source_repo="QuantStrategyLab/IssueRepo", dry_run=False, create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/issue/1", + comment_issue=lambda repo, url, body: "https://example.test/comment", list_issues=lambda repo: {}, ) @@ -65,11 +67,14 @@ def test_non_dry_run_skips_existing_open_issue(self) -> None: payload, dry_run=False, create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/new", + comment_issue=lambda repo, url, body: "https://example.test/comment", list_issues=lambda repo: {title: "https://example.test/existing"}, ) self.assertFalse(result["issues"][0]["created"]) self.assertEqual(result["issues"][0]["existing_url"], "https://example.test/existing") + self.assertEqual(result["issues"][0]["comment_url"], "https://example.test/comment") + self.assertTrue(result["issues"][0]["commented"]) self.assertEqual(calls, []) def test_resolve_input_path_rejects_metrics_path_traversal(self) -> None: @@ -127,7 +132,7 @@ def test_run_watcher_uses_validated_source_repo_in_task_summary(self) -> None: ) self.assertIn("QuantStrategyLab/TestStrategies:live", result["issues"][0]["title"]) - self.assertRegex(result["issues"][0]["title"], r"\[[a-f0-9]{12}\]$") + self.assertNotRegex(result["issues"][0]["title"], r"\[[a-f0-9]{12}\]$") self.assertEqual(result["issues"][0]["task"]["proposed_action"]["target"], "QuantStrategyLab/TestStrategies") def test_run_watcher_records_issue_errors_per_finding(self) -> None: @@ -142,7 +147,8 @@ def test_run_watcher_records_issue_errors_per_finding(self) -> None: result = run_watcher( payload, dry_run=False, - create_issue=lambda repo, title, body: (_ for _ in ()).throw(RuntimeError("boom")) if ":a " in title else "https://example.test/new", + create_issue=lambda repo, title, body: (_ for _ in ()).throw(RuntimeError("boom")) if ":a" in title else "https://example.test/new", + comment_issue=lambda repo, url, body: "https://example.test/comment", list_issues=lambda repo: {}, ) @@ -164,6 +170,7 @@ def test_run_watcher_updates_cache_after_create(self) -> None: payload, dry_run=False, create_issue=lambda repo, title, body: create_calls.append((repo, title, body)) or "https://example.test/new", + comment_issue=lambda repo, url, body: "https://example.test/comment", list_issues=lambda repo: {}, ) diff --git a/tests/test_strategy_watch.py b/tests/test_strategy_watch.py index 38372ca0..9db6732f 100644 --- a/tests/test_strategy_watch.py +++ b/tests/test_strategy_watch.py @@ -69,7 +69,7 @@ def test_issue_body_states_safety_boundary(self) -> None: issue = issue_for_task(finding_to_automation_task(finding)) self.assertIn("AI strategy optimization proposal", issue["title"]) - self.assertRegex(issue["title"], r"\[[a-f0-9]{12}\]$") + self.assertNotRegex(issue["title"], r"\[[a-f0-9]{12}\]$") self.assertIn("Event key", issue["body"]) self.assertIn("only opens an issue", issue["body"]) self.assertIn("does not modify strategy code", issue["body"]) From 748511040517b73a21c14432998a40fa44d5ca49 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 7 Jul 2026 02:18:31 +0800 Subject: [PATCH 9/9] fix: require validated watcher targets Co-Authored-By: Codex --- scripts/run_strategy_optimization_watcher.py | 23 ++++++++++------ service/automation_contracts.py | 7 ++++- service/strategy_watch.py | 10 +++++++ tests/test_automation_contracts.py | 16 +++++++++--- .../test_run_strategy_optimization_watcher.py | 26 +++++++++++++++---- tests/test_strategy_watch.py | 3 ++- 6 files changed, 66 insertions(+), 19 deletions(-) diff --git a/scripts/run_strategy_optimization_watcher.py b/scripts/run_strategy_optimization_watcher.py index 3e17368d..bf288eec 100755 --- a/scripts/run_strategy_optimization_watcher.py +++ b/scripts/run_strategy_optimization_watcher.py @@ -16,7 +16,7 @@ if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) -from service.strategy_watch import evaluate_strategy_watch, finding_to_automation_task, issue_for_task # noqa: E402 +from service.strategy_watch import evaluate_strategy_watch, finding_to_automation_task, issue_for_task, watcher_issue_key # noqa: E402 REPO_RE = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") @@ -93,16 +93,17 @@ def list_open_issue_urls(repo: str) -> dict[str, str]: for issue in issues: if not isinstance(issue, dict) or "pull_request" in issue: continue - title = str(issue.get("title") or "") - if title and title not in open_issues: - open_issues[title] = str(issue.get("html_url") or issue.get("url") or "") + body = str(issue.get("body") or "") + match = re.search(r"", body) + if match and match.group(1) not in open_issues: + open_issues[match.group(1)] = str(issue.get("html_url") or issue.get("url") or "") if len(issues) < 100: return open_issues page += 1 -def find_existing_open_issue(repo: str, title: str) -> str: - return list_open_issue_urls(repo).get(title, "") +def find_existing_open_issue(repo: str, issue_key: str) -> str: + return list_open_issue_urls(repo).get(issue_key, "") def task_public_summary(task: Any) -> dict[str, Any]: @@ -191,6 +192,10 @@ def run_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]: + if not dry_run and not source_repo: + raise ValueError("source_repo is required for non-dry-run strategy watcher runs") + if source_repo and not REPO_RE.fullmatch(source_repo): + raise ValueError("source_repo must be in owner/name form") watch_payload = _payload_for_source_repo(payload, source_repo) findings = evaluate_strategy_watch(watch_payload) issues: list[dict[str, Any]] = [] @@ -198,11 +203,13 @@ def run_watcher( for finding in findings: task = finding_to_automation_task(finding) issue = issue_for_task(task) + issue_key = watcher_issue_key(task) repo = source_repo or finding.snapshot.repo issue_result: dict[str, Any] = { "repo": repo, "title": issue["title"], "task": task_public_summary(task), + "watcher_issue_key": issue_key, "created": False, } if dry_run: @@ -211,7 +218,7 @@ def run_watcher( try: if repo not in open_issue_cache: open_issue_cache[repo] = list_issues(repo) - existing_url = open_issue_cache[repo].get(issue["title"], "") + existing_url = open_issue_cache[repo].get(issue_key, "") if existing_url: issue_result["existing_url"] = existing_url issue_result["comment_url"] = comment_issue(repo, existing_url, issue["body"]) @@ -219,7 +226,7 @@ def run_watcher( issue_result["skipped_reason"] = "open issue already exists; appended watcher update" else: issue_result["url"] = create_issue(repo, issue["title"], issue["body"]) - open_issue_cache[repo][issue["title"]] = str(issue_result["url"]) + open_issue_cache[repo][issue_key] = str(issue_result["url"]) issue_result["created"] = True except (ValueError, RuntimeError, subprocess.CalledProcessError) as exc: issue_result["error"] = str(exc) diff --git a/service/automation_contracts.py b/service/automation_contracts.py index 35aa492d..8fbb2242 100644 --- a/service/automation_contracts.py +++ b/service/automation_contracts.py @@ -109,7 +109,12 @@ def __post_init__(self) -> None: @property def is_actionable(self) -> bool: - return self.gate_decision.allowed and bool(self.proposed_action.action.strip()) + return ( + self.gate_decision.allowed + and bool(self.proposed_action.action.strip()) + and not self.gate_decision.human_review_required + and not self.proposed_action.requires_human_review + ) def to_dict(self) -> dict[str, Any]: return { diff --git a/service/strategy_watch.py b/service/strategy_watch.py index 4b742513..b0162a72 100644 --- a/service/strategy_watch.py +++ b/service/strategy_watch.py @@ -171,6 +171,14 @@ def finding_to_automation_task(finding: StrategyWatchFinding) -> AutomationTask: ) +def watcher_issue_key(task: AutomationTask) -> str: + payload = task.to_dict() + trigger = payload.get("trigger") if isinstance(payload.get("trigger"), dict) else {} + subject = str(trigger.get("subject") or "") + raw = json.dumps({"subject": subject}, ensure_ascii=False, sort_keys=True).encode("utf-8") + return hashlib.sha256(raw).hexdigest()[:16] + + def issue_for_task(task: AutomationTask) -> dict[str, str]: """Build a GitHub issue title/body for a strategy optimization task.""" payload = task.to_dict() @@ -178,12 +186,14 @@ def issue_for_task(task: AutomationTask) -> dict[str, str]: evidence = payload["evidence"] action = payload["proposed_action"] event_key = str(payload.get("metadata", {}).get("event_key") or "") + issue_key = watcher_issue_key(task) title = f"AI strategy optimization proposal: {trigger.get('subject') or action.get('target') or 'strategy profile'}" signals = "\n".join(f"- {item}" for item in trigger.get("evidence", [])) or "- Strategy metrics degraded." checks = "\n".join(f"- [ ] {item}" for item in payload["gate_decision"].get("required_checks", [])) risks = "\n".join(f"- {item}" for item in evidence.get("risks", [])) body = "\n".join( [ + f"", "## Summary", str(evidence.get("summary") or "Strategy optimization watcher opened this issue."), "", diff --git a/tests/test_automation_contracts.py b/tests/test_automation_contracts.py index 72409d45..1ecfa1ee 100644 --- a/tests/test_automation_contracts.py +++ b/tests/test_automation_contracts.py @@ -158,18 +158,26 @@ def test_is_actionable(self) -> None: allowed_task = AutomationTask( trigger=trigger, evidence=evidence, - proposed_action=ProposedAction("act", "lane", "target", "why"), - gate_decision=GateDecision(True, "ok"), + proposed_action=ProposedAction("act", "lane", "target", "why", requires_human_review=False), + gate_decision=GateDecision(True, "ok", human_review_required=False), ) blocked_task = AutomationTask( trigger=trigger, evidence=evidence, - proposed_action=ProposedAction("act", "lane", "target", "why"), - gate_decision=GateDecision(False, "blocked"), + proposed_action=ProposedAction("act", "lane", "target", "why", requires_human_review=False), + gate_decision=GateDecision(False, "blocked", human_review_required=False), + ) + + human_review_task = AutomationTask( + trigger=trigger, + evidence=evidence, + proposed_action=ProposedAction("act", "lane", "target", "why", requires_human_review=True), + gate_decision=GateDecision(True, "ok", human_review_required=True), ) self.assertTrue(allowed_task.is_actionable) self.assertFalse(blocked_task.is_actionable) + self.assertFalse(human_review_task.is_actionable) if __name__ == "__main__": diff --git a/tests/test_run_strategy_optimization_watcher.py b/tests/test_run_strategy_optimization_watcher.py index c8885a8b..88ef9f6b 100644 --- a/tests/test_run_strategy_optimization_watcher.py +++ b/tests/test_run_strategy_optimization_watcher.py @@ -61,14 +61,16 @@ def test_non_dry_run_skips_existing_open_issue(self) -> None: "current_metrics": {"sharpe": 0.5}, "baseline_metrics": {"sharpe": 1.0}, } - title = run_watcher(payload, dry_run=True)["issues"][0]["title"] + dry_result = run_watcher(payload, dry_run=True) + issue_key = dry_result["issues"][0]["watcher_issue_key"] result = run_watcher( payload, + source_repo="QuantStrategyLab/TestStrategies", dry_run=False, create_issue=lambda repo, title, body: calls.append((repo, title, body)) or "https://example.test/new", comment_issue=lambda repo, url, body: "https://example.test/comment", - list_issues=lambda repo: {title: "https://example.test/existing"}, + list_issues=lambda repo: {issue_key: "https://example.test/existing"}, ) self.assertFalse(result["issues"][0]["created"]) @@ -92,8 +94,8 @@ def test_resolve_input_path_accepts_source_relative_metrics_path(self) -> None: def test_find_existing_open_issue_paginates_until_exact_match(self) -> None: calls: list[list[str]] = [] - first_page = [{"title": f"other-{i}", "html_url": f"https://example.test/{i}"} for i in range(100)] - second_page = [{"title": "target", "html_url": "https://example.test/target"}] + first_page = [{"title": f"other-{i}", "body": "", "html_url": f"https://example.test/{i}"} for i in range(100)] + second_page = [{"title": "target", "body": "", "html_url": "https://example.test/target"}] def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: calls.append(cmd) @@ -104,11 +106,22 @@ def fake_run(cmd: list[str], **kwargs: object) -> subprocess.CompletedProcess[st with patch("scripts.run_strategy_optimization_watcher.subprocess.run", fake_run): issues = list_open_issue_urls("QuantStrategyLab/TestStrategies") - self.assertEqual(issues["target"], "https://example.test/target") + self.assertEqual(issues["abc12345"], "https://example.test/target") self.assertEqual(len(calls), 2) self.assertIn("--method", calls[0]) self.assertIn("GET", calls[0]) + def test_run_watcher_requires_source_repo_for_non_dry_run(self) -> None: + with self.assertRaises(ValueError): + run_watcher( + { + "repo": "QuantStrategyLab/TestStrategies", + "current_metrics": {"sharpe": 0.5}, + "baseline_metrics": {"sharpe": 1.0}, + }, + dry_run=False, + ) + def test_run_watcher_rejects_mismatched_source_repo_payload(self) -> None: with self.assertRaises(ValueError): run_watcher( @@ -146,6 +159,7 @@ def test_run_watcher_records_issue_errors_per_finding(self) -> None: result = run_watcher( payload, + source_repo="QuantStrategyLab/TestStrategies", dry_run=False, create_issue=lambda repo, title, body: (_ for _ in ()).throw(RuntimeError("boom")) if ":a" in title else "https://example.test/new", comment_issue=lambda repo, url, body: "https://example.test/comment", @@ -168,6 +182,7 @@ def test_run_watcher_updates_cache_after_create(self) -> None: result = run_watcher( payload, + source_repo="QuantStrategyLab/TestStrategies", dry_run=False, create_issue=lambda repo, title, body: create_calls.append((repo, title, body)) or "https://example.test/new", comment_issue=lambda repo, url, body: "https://example.test/comment", @@ -191,6 +206,7 @@ def test_run_watcher_caches_open_issues_per_repo(self) -> None: result = run_watcher( payload, + source_repo="QuantStrategyLab/TestStrategies", dry_run=False, create_issue=lambda repo, title, body: create_calls.append((repo, title, body)) or "https://example.test/new", list_issues=lambda repo: list_calls.append(repo) or {}, diff --git a/tests/test_strategy_watch.py b/tests/test_strategy_watch.py index 9db6732f..5a4b68b3 100644 --- a/tests/test_strategy_watch.py +++ b/tests/test_strategy_watch.py @@ -25,7 +25,7 @@ def test_degraded_snapshot_becomes_issue_only_task(self) -> None: self.assertEqual(len(findings), 1) task = finding_to_automation_task(findings[0]) payload = task.to_dict() - self.assertTrue(task.is_actionable) + self.assertFalse(task.is_actionable) self.assertEqual(payload["proposed_action"]["action"], "open_issue") self.assertTrue(payload["proposed_action"]["requires_human_review"]) self.assertTrue(payload["gate_decision"]["human_review_required"]) @@ -71,6 +71,7 @@ def test_issue_body_states_safety_boundary(self) -> None: self.assertIn("AI strategy optimization proposal", issue["title"]) self.assertNotRegex(issue["title"], r"\[[a-f0-9]{12}\]$") self.assertIn("Event key", issue["body"]) + self.assertIn("