diff --git a/.github/workflows/pr-agent.yml b/.github/workflows/pr-agent.yml new file mode 100644 index 0000000..c317051 --- /dev/null +++ b/.github/workflows/pr-agent.yml @@ -0,0 +1,64 @@ +--- +name: pr-agent + +# Advisory AI PR review (same loop as network-operations/noc-agent). Runs on +# the UNPRIVILEGED `hyrule-public-pr` runner only, with read + PR/issue-comment +# permissions and our own OpenRouter key. Never deploys, never writes code, +# never auto-merges. Config (model/fallback/instructions): .pr_agent.toml. +# Design: network-operations/docs/ci/pr-agent.md. +# +# Org prerequisites (one-time, done 2026-07-10): repo is in the public-pr +# runner group and in OPENROUTER_API_KEY's selected repositories. + +on: + pull_request: + types: [opened, synchronize, reopened, ready_for_review] + issue_comment: + types: [created] + +permissions: + contents: read + pull-requests: write + issues: write + +# Group by event name too: the push-triggered review posts comments, and each +# comment fires an issue_comment run that would otherwise share this group and +# cancel the in-flight check run ("higher priority waiting request", +# network-operations#408). Same-event storms still dedupe. +concurrency: + group: pr-agent-${{ github.event_name }}-${{ github.event.pull_request.number || github.event.issue.number }} + cancel-in-progress: true + +jobs: + pr-agent: + # Public-fork policy: auto-review ONLY for same-repo (internal) PRs, and + # slash commands ONLY from trusted authors. This keeps OPENROUTER_API_KEY + # off untrusted fork PRs and stops anyone from burning the OpenRouter + # budget via /ask, /improve, etc. + if: > + (github.event_name == 'pull_request' && + github.event.pull_request.head.repo.full_name == github.repository) || + (github.event_name == 'issue_comment' && + github.event.issue.pull_request && + startsWith(github.event.comment.body, '/') && + contains(fromJSON('["OWNER","MEMBER","COLLABORATOR"]'), github.event.comment.author_association)) + runs-on: [self-hosted, linux, x64, hyrule-public-pr] + timeout-minutes: 12 + steps: + - name: PR-Agent + uses: The-PR-Agent/pr-agent@8e4d32e5497defd43c023a404f73560c62728961 # v0.39.0 + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + # PR-Agent reads the OpenRouter key from openrouter__key; litellm also + # honours OPENROUTER_API_KEY. Set both from the org secret. + OPENROUTER__KEY: ${{ secrets.OPENROUTER_API_KEY }} + OPENROUTER_API_KEY: ${{ secrets.OPENROUTER_API_KEY }} + # Model pin via dynaconf env (SECTION__KEY). .pr_agent.toml is only + # honoured from the DEFAULT branch (no checkout step in the action), + # so until this PR merges the toml is invisible and PR-Agent would + # fall back to its packaged gpt-5.5 default. Keep the env pin even + # after the toml lands (belt and braces; custom models require + # custom_model_max_tokens). + CONFIG__MODEL: openrouter/deepseek/deepseek-v4-flash + CONFIG__FALLBACK_MODELS: '["openrouter/minimax/minimax-m2.7"]' + CONFIG__CUSTOM_MODEL_MAX_TOKENS: "128000" diff --git a/.pr_agent.toml b/.pr_agent.toml new file mode 100644 index 0000000..2089467 --- /dev/null +++ b/.pr_agent.toml @@ -0,0 +1,48 @@ +# PR-Agent (advisory) configuration for AS215932/engineering-loop. +# Replaces hosted Sourcery. Runs only on the unprivileged hyrule-public-pr +# runner with read/comment-only perms, on our OpenRouter key. +# Design: network-operations/docs/ci/pr-agent.md. + +[config] +git_provider = "github" +# deepseek-v4-flash / minimax-m2.7 are custom models -> need custom_model_max_tokens. +model = "openrouter/deepseek/deepseek-v4-flash" +fallback_models = ["openrouter/minimax/minimax-m2.7"] +custom_model_max_tokens = 128000 +temperature = 0.2 +use_repo_settings_file = true +publish_output = true + +[github_action_config] +auto_review = true +auto_describe = false +auto_improve = true + +[pr_reviewer] +require_security_review = true +require_tests_review = true +require_score_review = true +require_estimate_effort_to_review = true +extra_instructions = """ +This is engineering-loop: the autonomous change-implementation runtime +(LangGraph daemon + Reliability Governor) that turns approved GitHub issues +into draft PRs inside guarded worktrees. HIGHEST RISK is autonomy-boundary +erosion. Flag, with specifics: +- anything that could merge, push to protected branches, or apply production + changes (the loop stops at a DRAFT PR — that invariant is absolute); +- consumption of issues without the loop:approved label, Governor + tier/capability-envelope weakening, or auto-approval beyond Tier 0/1 without + the success-history evidence path; +- path-allowlist / denied_path_globs / denied_content_patterns bypasses in + policy or worktree handling; secrets reaching prompts, logs, or artifacts; +- LHP trust violations: issue prose treated as trusted instead of the + HMAC-fetched payload, pointer-hash checks skipped, callbacks setting + verified/resolved (NOC/SOC own closure); +- budget/backstop weakening (runs/cost per day, CI-refusal, singleton lock); +- unpinned GitHub Actions, broad GITHUB_TOKEN permissions, added + pull_request_target, or pull_request jobs on privileged runner labels. +Do not restate the diff; surface risk, invariant breaks, missing tests. +""" + +[pr_description] +publish_description_as_comment = true diff --git a/pyproject.toml b/pyproject.toml index f5c1fd2..1af74bc 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,4 +36,4 @@ python_version = "3.12" strict = true [tool.uv.sources] -agent-core = { git = "https://github.com/AS215932/agent-core", tag = "v0.7.0" } +agent-core = { git = "https://github.com/AS215932/agent-core", tag = "v0.8.0" } diff --git a/src/hyrule_engineering_loop/agent_core_trace.py b/src/hyrule_engineering_loop/agent_core_trace.py index 33e50f0..401ce8d 100644 --- a/src/hyrule_engineering_loop/agent_core_trace.py +++ b/src/hyrule_engineering_loop/agent_core_trace.py @@ -32,20 +32,23 @@ def enabled() -> bool: def _sink_from_env() -> Any: sink_mod = importlib.import_module("agent_core.tracing.sink") - path_configured = bool(os.environ.get(PATH_ENV, "").strip()) - collector_configured = bool(os.environ.get(COLLECTOR_URL_ENV, "").strip()) - if path_configured or collector_configured: - return sink_mod.sink_from_env(FLAG_ENV) - - original_path = os.environ.get(PATH_ENV) - os.environ[PATH_ENV] = _DEFAULT_PATH + overlays: dict[str, str] = {} + if not os.environ.get(PATH_ENV, "").strip() and not os.environ.get(COLLECTOR_URL_ENV, "").strip(): + overlays[PATH_ENV] = _DEFAULT_PATH + if not enabled() and _insight_records_enabled(): + # sink_from_env gates on the master trace flag; the insight flag is a + # complete opt-in of its own, so satisfy the gate for this build only. + overlays[FLAG_ENV] = "1" + originals = {key: os.environ.get(key) for key in overlays} + os.environ.update(overlays) try: return sink_mod.sink_from_env(FLAG_ENV) finally: - if original_path is None: - os.environ.pop(PATH_ENV, None) - else: - os.environ[PATH_ENV] = original_path + for key, original in originals.items(): + if original is None: + os.environ.pop(key, None) + else: + os.environ[key] = original def emit_loop_trace(state: Mapping[str, Any]) -> int: @@ -74,6 +77,103 @@ def emit_published_trace(state: Mapping[str, Any], pr_results: list[dict[str, An return emit_loop_trace({**dict(state), "pr_status": "pushed", "pr_results": pr_results}) +def emit_insight_decision_envelopes( + insights: list[dict[str, Any]], + *, + input_event: Mapping[str, Any] | None = None, +) -> int: + """Emit one LoopDecisionEnvelope TraceEvent per insight record. + + Mirrors the NOC/SOC modules: the payload carries both the envelope and the + full validated ``InsightDecisionRecord`` (the envelope alone drops + sampling_class/utility/cost/support_facts, which the knowledge repo's + IDQ/CGS evaluation needs). Best-effort like everything else here. + """ + if not insights or not (enabled() or _insight_records_enabled()): + return 0 + try: + sink = _sink_from_env() + count = 0 + for insight in insights: + event = _insight_decision_event(insight, input_event=dict(input_event or {})) + if sink.emit(event): + count += 1 + return count + except Exception: + return 0 + + +def _insight_records_enabled() -> bool: + # HYRULE_ENGINEERING_INSIGHT_RECORDS=1 is a complete opt-in for insight + # envelopes: operators must not need to discover the trace master flag + # too, or the ledger and the trace sink silently diverge. The sink still + # honours the {FLAG_ENV}_COLLECTOR_URL/_PATH vars (JSONL fallback when + # neither is set). + return os.environ.get("HYRULE_ENGINEERING_INSIGHT_RECORDS", "").strip().lower() in _TRUTHY + + +def _insight_decision_event(insight: Mapping[str, Any], *, input_event: dict[str, Any]) -> Any: + contracts = importlib.import_module("agent_core.contracts") + TraceEvent = getattr(contracts, "TraceEvent") + LoopDecisionEnvelope = getattr(contracts, "LoopDecisionEnvelope") + InsightDecisionRecord = getattr(contracts, "InsightDecisionRecord") + + validated = InsightDecisionRecord.model_validate(dict(insight)) + # Correlate with the daemon/governor cycle when the insight carries no + # explicit trace id, so consumers can join the stream back to its run. + trace_id = validated.trace_id or _string_or_none(input_event.get("run_id")) + envelope = LoopDecisionEnvelope( + envelope_id=( + f"ldec_eng_{_stable_hash([validated.insight_id, validated.fingerprint, validated.action_selected])}" + ), + loop="engineering", + environment="production", + graph_id="engineering-loop", + node_id="insight_stream", + agent_role="engineering_loop", + run_id=_string_or_none(input_event.get("run_id")) or validated.run_id, + trace_id=trace_id, + input_event={ + **input_event, + "candidate_type": validated.candidate_type, + "candidate_source": validated.candidate_source, + }, + retrieved_context=validated.evidence_refs, + decision=validated.action_selected, + evidence_refs=validated.evidence_refs, + proposed_action={ + "candidate_type": validated.candidate_type, + "candidate_source": validated.candidate_source, + "why_now": validated.why_now, + "support_fact_count": len(validated.support_facts), + }, + human_outcome=validated.human_feedback, + governance=validated.governance, + insight_id=validated.insight_id, + case_id=validated.case_id, + fingerprint=validated.fingerprint, + policy_version=validated.policy_version, + ) + return TraceEvent( + event_type="loop_decision_envelope", + graph_id="engineering-loop", + node_id="loop_decision_envelope", + agent_role="engineering_loop", + environment="production", + run_id=envelope.run_id, + trace_id=envelope.trace_id, + summary=f"Engineering loop decision envelope for {validated.insight_id}", + payload={ + "loop_decision_envelope": envelope.model_dump(mode="json"), + "insight_decision_record": validated.model_dump(mode="json"), + # support_facts carry issue/signal titles — LHP-untrusted text — + # so the payload gets the same guard fields as the loop traces. + "untrusted_loop_text": True, + "model_consumption_allowed": False, + }, + ) + + def _trace_payload(state: Mapping[str, Any]) -> dict[str, Any]: payload = dict(state) pr_results = state.get("pr_results") diff --git a/src/hyrule_engineering_loop/capability_history.py b/src/hyrule_engineering_loop/capability_history.py new file mode 100644 index 0000000..4629d98 --- /dev/null +++ b/src/hyrule_engineering_loop/capability_history.py @@ -0,0 +1,258 @@ +"""Trustworthy success history for capability-envelope auto-approval. + +``decide_policy`` already contains the dormant Tier-2 branch: a capability +auto-approves tier-2 work iff its registry entry allows it AND +``success_count >= DEFAULT_STRONG_HISTORY_SUCCESSES`` with zero failures. +Every registry entry ships ``success_count: 0``, so the branch never fires. + +This module builds the evidence that could justify raising those numbers: +it joins stored CandidateDecisionRecords (the governor's audit trail) with +the terminal state of the pull requests that carried out the work (via the +same ``gh`` CLI client the intake uses) and emits a per-capability report +plus a proposed registry patch. The patch is NEVER applied automatically — +registry raises land as reviewed promotion PRs against the deployed registry +in network-operations, stamped policy_version +``reliability-governor.tier2-history.v1``. + +Outcome rules (90-day window by default): +- success: a PR whose body closes the issue was merged (branch protection + means required checks passed) by a human (non-bot login), with no revert + PR referencing it afterwards. +- failure: the PR was closed unmerged, or a revert referencing it merged. +- pending/none: open PRs and issues without PRs are excluded from counts. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +from hyrule_engineering_loop.governor import CandidateDecisionRecord +from hyrule_engineering_loop.intake import GhClient + +HISTORY_POLICY_VERSION = "reliability-governor.tier2-history.v1" +DEFAULT_WINDOW_DAYS = 90 + + +@dataclass(frozen=True) +class CapabilityOutcome: + capability: str + issue_id: str + pr_url: str + outcome: str # success | failure | pending | none + reason: str + risk_tier: int = 0 + + +@dataclass +class CapabilityHistory: + window_days: int + generated_at: str + outcomes: list[CapabilityOutcome] = field(default_factory=list) + + def aggregates(self) -> dict[str, dict[str, int]]: + totals: dict[str, dict[str, int]] = {} + for outcome in self.outcomes: + bucket = totals.setdefault(outcome.capability, {"success": 0, "failure": 0, "pending": 0, "none": 0}) + bucket[outcome.outcome] = bucket.get(outcome.outcome, 0) + 1 + return dict(sorted(totals.items())) + + def registry_proposal(self) -> dict[str, Any]: + """Proposed counts per capability (report artifact). + + ``success_count`` counts only tier>=2 successes — the numbers exist to + unlock ``decide_policy``'s Tier-2 gate, so a streak of human-merged + Tier-0/1 work must not satisfy it. Lower-tier successes and + pending/unknown outcomes are reported alongside so the promotion-PR + reviewer sees exactly what the evidence covers (and what it doesn't). + """ + proposal: dict[str, Any] = {} + for capability in sorted({outcome.capability for outcome in self.outcomes}): + rows = [outcome for outcome in self.outcomes if outcome.capability == capability] + proposal[capability] = { + "success_count": sum( + 1 for row in rows if row.outcome == "success" and row.risk_tier >= 2 + ), + "failure_count": sum(1 for row in rows if row.outcome == "failure"), + "lower_tier_success_count": sum( + 1 for row in rows if row.outcome == "success" and row.risk_tier < 2 + ), + "pending_count": sum(1 for row in rows if row.outcome == "pending"), + } + return proposal + + def as_dict(self) -> dict[str, Any]: + return { + "policy_version": HISTORY_POLICY_VERSION, + "window_days": self.window_days, + "generated_at": self.generated_at, + "aggregates": self.aggregates(), + "registry_proposal": self.registry_proposal(), + "evidence": [ + { + "capability": item.capability, + "issue_id": item.issue_id, + "pr_url": item.pr_url, + "outcome": item.outcome, + "reason": item.reason, + "risk_tier": item.risk_tier, + } + for item in self.outcomes + ], + } + + +def load_decision_records(state_dir: Path, *, window_days: int = DEFAULT_WINDOW_DAYS, now: datetime | None = None) -> list[CandidateDecisionRecord]: + """Stored governor decision records inside the evidence window, one per + (repo, issue): the newest record wins (label transitions re-record).""" + root = state_dir.expanduser().resolve() + if not root.exists(): + return [] + now = now or datetime.now(UTC) + cutoff = now - timedelta(days=window_days) + newest: dict[str, CandidateDecisionRecord] = {} + for path in sorted(root.glob("*.json")): + try: + record = CandidateDecisionRecord.model_validate_json(path.read_text(encoding="utf-8")) + except (OSError, ValueError): + continue + created = _parse_ts(record.created_at) + if created is None or created < cutoff: + continue + key = record.issue_id + prior = newest.get(key) + if prior is None or str(record.created_at) > str(prior.created_at): + newest[key] = record + return [newest[key] for key in sorted(newest)] + + +def pr_outcome_for_issue(client: GhClient, *, repo: str, issue_url: str) -> tuple[str, str, str]: + """(outcome, pr_url, reason) for the newest PR CLOSING this issue. + + The search requires the closing keyword the daemon writes verbatim + (``Closes ``), so PRs that merely mention the issue as related + context are never counted as its outcome.""" + try: + raw = client.run( + [ + "pr", + "list", + "--repo", + repo, + "--state", + "all", + "--search", + f'"Closes {issue_url}" in:body', + "--json", + "number,state,url,mergedAt,mergedBy,title", + "--limit", + "10", + ] + ) + rows = json.loads(raw or "[]") + except Exception as exc: + # Unknown, not absent: a rate-limited/auth-failed lookup must stay + # visible in the report instead of vanishing from the counts. + return "pending", "", f"pr lookup failed ({exc.__class__.__name__}); outcome unknown" + if not isinstance(rows, list) or not rows: + return "none", "", "no PR closes the issue" + rows.sort(key=lambda row: int(row.get("number") or 0), reverse=True) + pr = rows[0] + pr_url = str(pr.get("url") or "") + state = str(pr.get("state") or "").upper() + if state == "OPEN": + return "pending", pr_url, "PR still open" + if state == "CLOSED" and not pr.get("mergedAt"): + return "failure", pr_url, "PR closed without merge" + merged_by = str((pr.get("mergedBy") or {}).get("login") or "") + if not merged_by: + # mergedBy is nullable (deleted/unavailable actor) — no evidence of a + # human merge, so this must not count toward zero-failure history. + return "pending", pr_url, "merger unknown (mergedBy null); needs human-merge evidence" + if merged_by.endswith("[bot]"): + return "pending", pr_url, f"merged by bot ({merged_by}); needs human-merge evidence" + reverted = _was_reverted(client, repo=repo, pr_number=int(pr.get("number") or 0)) + if reverted is None: + # Unknown revert state must never manufacture a success for a + # zero-failure history gate. + return "pending", pr_url, "revert lookup failed; outcome unknown" + if reverted: + return "failure", pr_url, "a merged revert references this PR" + return "success", pr_url, f"merged by {merged_by or 'human'} behind required checks" + + +def _was_reverted(client: GhClient, *, repo: str, pr_number: int) -> bool | None: + """True/False when determinable, None when the lookup itself failed. + + Covers both revert forms: GitHub's revert button writes ``Reverts + /#N`` into the BODY (title is ``Revert ""``), + while hand-written reverts commonly carry ``Revert #N`` in the title.""" + if not pr_number: + return False + for search in (f'"Reverts {repo}#{pr_number}" in:body', f'"Revert #{pr_number}" in:title'): + try: + raw = client.run( + [ + "pr", + "list", + "--repo", + repo, + "--state", + "merged", + "--search", + search, + "--json", + "number", + "--limit", + "3", + ] + ) + rows = json.loads(raw or "[]") + except Exception: + return None + if isinstance(rows, list) and rows: + return True + return False + + +def build_capability_history( + state_dir: Path, + *, + client: GhClient, + window_days: int = DEFAULT_WINDOW_DAYS, + now: datetime | None = None, +) -> CapabilityHistory: + now = now or datetime.now(UTC) + history = CapabilityHistory(window_days=window_days, generated_at=now.isoformat()) + for record in load_decision_records(state_dir, window_days=window_days, now=now): + if not record.matched_capability: + continue + issue_url = f"https://github.com/{record.repo}/issues/{record.issue_number}" + outcome, pr_url, reason = pr_outcome_for_issue(client, repo=record.repo, issue_url=issue_url) + history.outcomes.append( + CapabilityOutcome( + capability=record.matched_capability, + issue_id=record.issue_id, + pr_url=pr_url, + outcome=outcome, + reason=reason, + risk_tier=int(record.risk_tier), + ) + ) + return history + + +def _parse_ts(value: str) -> datetime | None: + text = str(value or "").strip() + if not text: + return None + if text.endswith("Z"): + text = f"{text[:-1]}+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + return parsed if parsed.tzinfo else parsed.replace(tzinfo=UTC) diff --git a/src/hyrule_engineering_loop/cli.py b/src/hyrule_engineering_loop/cli.py index d64df56..8e12635 100644 --- a/src/hyrule_engineering_loop/cli.py +++ b/src/hyrule_engineering_loop/cli.py @@ -21,6 +21,12 @@ ) from hyrule_engineering_loop.agent_core_trace import emit_published_trace from hyrule_engineering_loop.graph import build_graph +from hyrule_engineering_loop.insights import ( + daemon_report_insight, + governor_report_insights, + intake_report_insights, + record_insights, +) from hyrule_engineering_loop.governor import ( ReliabilityGovernorConfig, reliability_governor_once, @@ -487,6 +493,7 @@ def backend_canary_command(args: argparse.Namespace) -> int: DEFAULT_INTAKE_REPO = "AS215932/network-operations" DEFAULT_INTAKE_REPOS = list(CORE_REPOS) +INTAKE_INSIGHT_STATE_DIR = Path(".engineering-loop-state/intake") def daemon_command(args: argparse.Namespace) -> int: @@ -520,6 +527,13 @@ def daemon_command(args: argparse.Namespace) -> int: knowledge_learning_dir=args.knowledge_learning_dir, ) report = daemon_once(config, client=GhCli()) + insight = daemon_report_insight(report.as_dict()) + if insight is not None: + record_insights( + [insight], + config.state_dir, + input_event={"run_id": report.change_id or "", "component": "engineering_daemon"}, + ) print(json.dumps(report.as_dict(), indent=2, sort_keys=True)) return 0 if report.outcome not in {"error", "refused_ci"} else 1 @@ -542,10 +556,35 @@ def governor_command(args: argparse.Namespace) -> int: dry_run=args.dry_run, ) report = reliability_governor_once(config, client=GhCli()) + record_insights( + governor_report_insights(report, dry_run=config.dry_run), + config.state_dir, + input_event={"component": "reliability_governor"}, + ) print(json.dumps(report.as_dict(), indent=2, sort_keys=True)) return 0 +def capability_history_command(args: argparse.Namespace) -> int: + """Build the evidence report behind capability-registry success counts.""" + from hyrule_engineering_loop.capability_history import build_capability_history + + history = build_capability_history( + Path(args.state_dir_path).expanduser() + if args.state_dir_path + else ReliabilityGovernorConfig.state_dir, + client=GhCli(), + window_days=args.window_days, + ) + payload = history.as_dict() + if args.write_proposal: + Path(args.write_proposal).write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8" + ) + print(json.dumps(payload, indent=2, sort_keys=True)) + return 0 + + def intake_scan_command(args: argparse.Namespace) -> int: client = GhCli() signals, skipped = mine_all_signals(repo=args.repo, client=client) @@ -553,6 +592,11 @@ def intake_scan_command(args: argparse.Namespace) -> int: signals, repo=args.repo, client=client, dry_run=args.dry_run ) report.skipped_miners = skipped + record_insights( + intake_report_insights(report, signals, repo=args.repo, dry_run=args.dry_run), + INTAKE_INSIGHT_STATE_DIR, + input_event={"component": "engineering_intake", "repo": args.repo}, + ) payload = report.as_dict() if args.json: print(json.dumps(payload, indent=2, sort_keys=True)) @@ -979,6 +1023,17 @@ def build_parser() -> argparse.ArgumentParser: help_text="deprecated alias for reliability-governor", ) + capability_history_parser = subparsers.add_parser( + "capability-history", + help="evidence report behind capability-registry success counts (Tier-2 auto-approval)", + ) + capability_history_parser.add_argument("--state-dir-path", dest="state_dir_path") + capability_history_parser.add_argument("--window-days", type=int, default=90) + capability_history_parser.add_argument( + "--write-proposal", help="also write the report JSON (evidence for a registry promotion PR)" + ) + capability_history_parser.set_defaults(func=capability_history_command) + intake_parser = subparsers.add_parser("intake", help="signal mining and triage inbox") intake_subparsers = intake_parser.add_subparsers(dest="intake_command", required=True) diff --git a/src/hyrule_engineering_loop/insights.py b/src/hyrule_engineering_loop/insights.py index bd5426c..3b2c19a 100644 --- a/src/hyrule_engineering_loop/insights.py +++ b/src/hyrule_engineering_loop/insights.py @@ -9,6 +9,7 @@ import hashlib import importlib import json +import os from datetime import UTC, datetime from pathlib import Path from typing import Any, Literal @@ -16,6 +17,44 @@ InsightAction = Literal["notify", "question", "draft", "stay_silent"] SamplingClass = Literal["surfaced", "withheld_logged", "sampled_quiet_interval"] +# Master switch for production insight recording (ledger + envelope emission). +# The builders below stay pure/testable; only record_insights() consults this. +INSIGHT_RECORDS_ENV = "HYRULE_ENGINEERING_INSIGHT_RECORDS" +_TRUTHY = {"1", "true", "yes", "on"} + + +def insight_records_enabled() -> bool: + return os.environ.get(INSIGHT_RECORDS_ENV, "").strip().lower() in _TRUTHY + + +def record_insights( + records: list[dict[str, Any]], + state_dir: Path, + *, + input_event: dict[str, Any] | None = None, +) -> int: + """Persist records to the private ledger and emit decision envelopes. + + Flag-gated by ``HYRULE_ENGINEERING_INSIGHT_RECORDS`` and strictly + best-effort: a ledger or delivery failure only loses observability. + Returns the count delivered to at least one sink. + """ + if not insight_records_enabled() or not records: + return 0 + for record in records: + try: + write_insight_record(record, state_dir) + except Exception: # noqa: BLE001 - one bad record must not drop the rest + continue + try: + from hyrule_engineering_loop import agent_core_trace + + return agent_core_trace.emit_insight_decision_envelopes( + records, input_event=input_event or {} + ) + except Exception: # noqa: BLE001 + return 0 + def write_insight_record(record: dict[str, Any], state_dir: Path) -> Path: _validate_with_agent_core(record) @@ -61,6 +100,10 @@ def signal_insight_record( ) -> dict[str, Any]: utility = _utility_for_signal(action_items=action_items, related=related, issue_ref=issue_ref) cost = _interruption_cost_for_action(action_selected=action_selected, why_now=why_now) + # The filed/deduped issue is the record's primary evidence — envelopes are + # built from evidence_refs, so it must live there (budget_context alone is + # not joinable downstream). + issue_refs = [{"kind": "github_issue", "ref": issue_ref}] if issue_ref else [] return _base_record( fingerprint=fingerprint, sampling_class=sampling_class, @@ -69,7 +112,11 @@ def signal_insight_record( action_selected=action_selected, why_now=why_now, support_facts=[title, context, *list(action_items)[:4]], - evidence_refs=[{"kind": "repo", "ref": repo}, *[{"kind": "related", "ref": item} for item in list(related)[:6]]], + evidence_refs=[ + *issue_refs, + {"kind": "repo", "ref": repo}, + *[{"kind": "related", "ref": item} for item in list(related)[:6]], + ], expected_utility=utility, interruption_cost=cost, confidence=0.75 if action_selected != "stay_silent" else 0.65, @@ -88,9 +135,16 @@ def governor_insight_record( policy_version: str, action_selected: InsightAction | None = None, sampling_class: SamplingClass = "surfaced", + knowledge_context_pack_id: str = "", + knowledge_export_version: str = "", ) -> dict[str, Any]: action = action_selected or _action_for_routing_decision(routing_decision) why_now = "; ".join(reasons[:3]) or f"Governor routed issue as {routing_decision}." + knowledge_refs = ( + [{"kind": "okf_context_pack", "ref": knowledge_context_pack_id}] + if knowledge_context_pack_id + else [] + ) return _base_record( fingerprint=_stable_hash(f"governor:{issue_id}:{routing_decision}"), sampling_class=sampling_class, @@ -99,7 +153,14 @@ def governor_insight_record( action_selected=action, why_now=why_now, support_facts=[title, routing_decision, *reasons[:6]], - evidence_refs=[{"kind": "github_issue", "ref": issue_id}, *[{"kind": "label", "ref": label} for label in labels]], + evidence_refs=[ + {"kind": "github_issue", "ref": issue_id}, + *knowledge_refs, + *[{"kind": "label", "ref": label} for label in labels], + ], + tool_versions=( + {"knowledge_export": knowledge_export_version} if knowledge_export_version else {} + ), expected_utility={ "total": 0.7 if action != "stay_silent" else 0.15, "components": {"routing_decision": 0.7 if action != "stay_silent" else 0.15}, @@ -157,6 +218,7 @@ def _base_record( confidence: float, policy_version: str, budget_context: dict[str, Any], + tool_versions: dict[str, str] | None = None, ) -> dict[str, Any]: now = datetime.now(UTC).isoformat() return { @@ -180,7 +242,7 @@ def _base_record( "confidence": confidence, "risk_class": "medium", "policy_version": policy_version, - "tool_versions": {}, + "tool_versions": tool_versions or {}, "budget_context": budget_context, } @@ -237,3 +299,126 @@ def _why_not_other_actions(action_selected: InsightAction) -> dict[str, str]: def _stable_hash(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest()[:16] + + +# --- report -> insight mappings (called from the CLI entry points) ----------- + +_DAEMON_ACTIONS: dict[str, tuple[InsightAction, SamplingClass]] = { + "published": ("draft", "surfaced"), + "needs_triage": ("question", "surfaced"), + "idle": ("stay_silent", "sampled_quiet_interval"), + "over_budget": ("stay_silent", "withheld_logged"), + "refused_ci": ("stay_silent", "withheld_logged"), + "error": ("stay_silent", "withheld_logged"), +} + + +def daemon_report_insight(report: dict[str, Any]) -> dict[str, Any] | None: + """One insight per daemon cycle; ``locked``/unknown outcomes emit nothing + (another cycle is already reporting).""" + outcome = str(report.get("outcome") or "") + mapped = _DAEMON_ACTIONS.get(outcome) + if mapped is None: + return None + action, sampling = mapped + issue = report.get("issue") or {} + issue_ref = str(issue.get("url") or issue.get("issue_id") or "") + if not issue_ref and issue.get("repo") and issue.get("number"): + # daemon_once reports carry {repo, number, title} only — build the + # structural ref so the record stays joinable to the GitHub issue. + issue_ref = f"https://github.com/{issue['repo']}/issues/{issue['number']}" + why_now = f"daemon cycle {outcome}" + (f": {report['detail']}" if report.get("detail") else "") + budget_context: dict[str, Any] = { + "outcome": outcome, + "cost_usd": report.get("cost_usd", 0.0), + } + if report.get("pr_url"): + budget_context["pr_url"] = report["pr_url"] + return daemon_insight_record( + action_selected=action, + sampling_class=sampling, + why_now=why_now[:300], + issue_ref=issue_ref, + budget_context=budget_context, + ) + + +def governor_report_insights(report: Any, *, dry_run: bool) -> list[dict[str, Any]]: + """One insight per fresh Reliability Governor decision. + + ``skipped`` entries (": unchanged decision ...") are the + governor's dedup — an unchanged decision is not a new insight. + """ + skipped_ids = {str(entry).split(":", 1)[0] for entry in getattr(report, "skipped", [])} + records: list[dict[str, Any]] = [] + for decision in getattr(report, "records", []): + issue_id = str(getattr(decision, "issue_id", "")) + if issue_id in skipped_ids: + continue + reasons = list(getattr(decision, "denial_reasons", []) or []) or list( + getattr(decision, "policy_rules", []) or [] + ) + # Issue prose is untrusted (LHP rule); identify the issue structurally. + title = f"{getattr(decision, 'repo', '')}#{getattr(decision, 'issue_number', '')} {getattr(decision, 'intent_type', '')}".strip() + records.append( + governor_insight_record( + issue_id=issue_id, + title=title, + routing_decision=str(getattr(decision, "routing_decision", "")), + reasons=[str(reason) for reason in reasons], + labels=[str(label) for label in getattr(decision, "labels_to_add", []) or []], + policy_version=str(getattr(decision, "schema_version", "governor")), + sampling_class="withheld_logged" if dry_run else "surfaced", + knowledge_context_pack_id=str(getattr(decision, "knowledge_context_pack_id", "")), + knowledge_export_version=str(getattr(decision, "knowledge_export_version", "")), + ) + ) + return records + + +def intake_report_insights( + report: Any, signals: list[Any], *, repo: str, dry_run: bool +) -> list[dict[str, Any]]: + """Filed signals surface as drafts; fingerprint-deduped ones are explicit + silence (the open issue already carries the information).""" + by_fingerprint = {getattr(signal, "fingerprint", ""): signal for signal in signals} + records: list[dict[str, Any]] = [] + for entry in getattr(report, "filed", []): + signal = by_fingerprint.get(str(entry.get("fingerprint") or "")) + records.append( + signal_insight_record( + repo=repo, + source=str(entry.get("source") or "intake"), + identifier=str(entry.get("fingerprint") or ""), + title=str(entry.get("title") or ""), + context=str(getattr(signal, "context", "") or "")[:300], + action_items=list(getattr(signal, "action_items", []) or []), + related=list(getattr(signal, "related", []) or []), + fingerprint=str(entry.get("fingerprint") or ""), + action_selected="draft", + sampling_class="withheld_logged" if dry_run else "surfaced", + why_now="new mined signal filed as loop:candidate" + + (" (dry-run, nothing filed)" if dry_run else ""), + issue_ref=str(entry.get("url") or ""), + ) + ) + for entry in getattr(report, "deduplicated", []): + signal = by_fingerprint.get(str(entry.get("fingerprint") or "")) + existing = str(entry.get("existing_issue") or "") + records.append( + signal_insight_record( + repo=repo, + source=str(getattr(signal, "source", "") or "intake"), + identifier=str(entry.get("fingerprint") or ""), + title=str(entry.get("title") or ""), + context="", + action_items=[], + related=[], + fingerprint=str(entry.get("fingerprint") or ""), + action_selected="stay_silent", + sampling_class="withheld_logged", + why_now=f"fingerprint dedupe: open issue #{existing}", + issue_ref=f"https://github.com/{repo}/issues/{existing}" if existing else "", + ) + ) + return records diff --git a/tests/test_capability_history.py b/tests/test_capability_history.py new file mode 100644 index 0000000..f08a1d3 --- /dev/null +++ b/tests/test_capability_history.py @@ -0,0 +1,249 @@ +"""Capability success-history evidence (Tier-2 auto-approval prerequisite).""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime +from pathlib import Path + +from hyrule_engineering_loop.capability_history import ( + build_capability_history, + load_decision_records, + pr_outcome_for_issue, +) +from hyrule_engineering_loop.governor import CandidateDecisionRecord, write_decision_record + +NOW = datetime(2026, 7, 10, 12, 0, 0, tzinfo=UTC) + + +def _record(issue_number: int, *, capability: str | None = "tier0.docs-runbooks-tests", created_at: str | None = None) -> CandidateDecisionRecord: + return CandidateDecisionRecord( + record_id=f"rdr_{issue_number}", + created_at=created_at or "2026-07-01T00:00:00Z", + issue_id=f"AS215932/network-operations#{issue_number}", + repo="AS215932/network-operations", + issue_number=issue_number, + authority_text_hash="hash", + issue_text_hash="hash", + source="noc", + intent_type="docs", + risk_tier=0, + blast_radius="docs-only", + affected_assets=[], + affected_services=[], + affected_customers=[], + knowledge_export_version="run:retr:pol", + knowledge_context_pack_id="ctx", + knowledge_authority_level="A1", + knowledge_status="current", + matched_capability=capability, + routing_decision="allow_approved", + next_loop="engineering", + handoff_contract="github_issue_labels", + labels_to_add=["loop:approved"], + ) + + +class FakeGh: + def __init__(self, responses: dict[str, str]): + self.responses = responses + self.calls: list[list[str]] = [] + + def run(self, args: list[str]) -> str: + self.calls.append(args) + search = "" + for index, token in enumerate(args): + if token == "--search": + search = args[index + 1] + for needle, response in self.responses.items(): + if needle in search: + return response + return "[]" + + +def _merged_pr(number: int, *, merged_by: str = "svag") -> str: + return json.dumps( + [ + { + "number": number, + "state": "MERGED", + "url": f"https://github.com/AS215932/network-operations/pull/{number}", + "mergedAt": "2026-07-02T00:00:00Z", + "mergedBy": {"login": merged_by}, + "title": "docs: fix", + } + ] + ) + + +def test_load_decision_records_window_and_newest_wins(tmp_path: Path) -> None: + old = _record(1, created_at="2026-01-01T00:00:00Z") + fresh_v1 = _record(2, created_at="2026-07-01T00:00:00Z") + fresh_v2 = _record(2, created_at="2026-07-05T00:00:00Z") + for record in (old, fresh_v1, fresh_v2): + write_decision_record(record, tmp_path) + records = load_decision_records(tmp_path, window_days=90, now=NOW) + assert [record.issue_number for record in records] == [2] + assert records[0].created_at == "2026-07-05T00:00:00Z" + + +def test_pr_outcomes() -> None: + issue_url = "https://github.com/AS215932/network-operations/issues/2" + merged = FakeGh({issue_url: _merged_pr(7), "Revert #7": "[]"}) + assert pr_outcome_for_issue(merged, repo="AS215932/network-operations", issue_url=issue_url)[0] == "success" + + reverted = FakeGh({issue_url: _merged_pr(7), "Revert #7": json.dumps([{"number": 9}])}) + assert pr_outcome_for_issue(reverted, repo="AS215932/network-operations", issue_url=issue_url)[0] == "failure" + + closed = FakeGh( + {issue_url: json.dumps([{"number": 7, "state": "CLOSED", "url": "u", "mergedAt": None}])} + ) + assert pr_outcome_for_issue(closed, repo="AS215932/network-operations", issue_url=issue_url)[0] == "failure" + + bot_merged = FakeGh({issue_url: _merged_pr(7, merged_by="some-bot[bot]")}) + assert pr_outcome_for_issue(bot_merged, repo="AS215932/network-operations", issue_url=issue_url)[0] == "pending" + + none = FakeGh({}) + assert pr_outcome_for_issue(none, repo="AS215932/network-operations", issue_url=issue_url)[0] == "none" + + +def test_build_history_aggregates_and_proposal(tmp_path: Path) -> None: + write_decision_record(_record(2, created_at="2026-07-01T00:00:00Z"), tmp_path) + write_decision_record(_record(3, created_at="2026-07-02T00:00:00Z"), tmp_path) + write_decision_record(_record(4, created_at="2026-07-03T00:00:00Z", capability=None), tmp_path) + gh = FakeGh( + { + "issues/2": _merged_pr(7), + "issues/3": json.dumps([{"number": 8, "state": "CLOSED", "url": "u8", "mergedAt": None}]), + } + ) + history = build_capability_history(tmp_path, client=gh, window_days=90, now=NOW) + # the record without a matched capability contributes nothing + assert len(history.outcomes) == 2 + aggregates = history.aggregates()["tier0.docs-runbooks-tests"] + assert aggregates["success"] == 1 + assert aggregates["failure"] == 1 + proposal = history.registry_proposal()["tier0.docs-runbooks-tests"] + # both records are tier-0, so the merged one is a lower-tier success and + # the Tier-2 gate count stays untouched + assert proposal == { + "success_count": 0, + "failure_count": 1, + "lower_tier_success_count": 1, + "pending_count": 0, + } + payload = history.as_dict() + assert payload["policy_version"] == "reliability-governor.tier2-history.v1" + assert len(payload["evidence"]) == 2 + + +class RaisingRevertGh(FakeGh): + def run(self, args: list[str]) -> str: + joined = " ".join(args) + if "Revert" in joined: + raise RuntimeError("rate limited") + return super().run(args) + + +def test_outcome_search_requires_closing_keyword() -> None: + issue_url = "https://github.com/AS215932/network-operations/issues/2" + gh = FakeGh({issue_url: _merged_pr(7)}) + pr_outcome_for_issue(gh, repo="AS215932/network-operations", issue_url=issue_url) + searches = [args[args.index("--search") + 1] for args in gh.calls if "--search" in args] + # the PR lookup must require the closing keyword, not a bare URL mention + assert searches[0] == f'"Closes {issue_url}" in:body' + + +def test_revert_detected_in_github_button_body_form() -> None: + issue_url = "https://github.com/AS215932/network-operations/issues/2" + gh = FakeGh( + { + issue_url: _merged_pr(7), + "Reverts AS215932/network-operations#7": json.dumps([{"number": 9}]), + } + ) + outcome, _url, reason = pr_outcome_for_issue( + gh, repo="AS215932/network-operations", issue_url=issue_url + ) + assert outcome == "failure" + assert "revert" in reason + + +def test_revert_lookup_failure_is_unknown_not_success() -> None: + issue_url = "https://github.com/AS215932/network-operations/issues/2" + gh = RaisingRevertGh({issue_url: _merged_pr(7)}) + outcome, _url, reason = pr_outcome_for_issue( + gh, repo="AS215932/network-operations", issue_url=issue_url + ) + assert outcome == "pending" + assert "unknown" in reason + + +def test_newest_pr_sorts_numerically_not_lexicographically() -> None: + issue_url = "https://github.com/AS215932/network-operations/issues/2" + closed_99_merged_100 = json.dumps( + [ + {"number": 99, "state": "CLOSED", "url": "u99", "mergedAt": None}, + { + "number": 100, + "state": "MERGED", + "url": "u100", + "mergedAt": "2026-07-02T00:00:00Z", + "mergedBy": {"login": "svag"}, + }, + ] + ) + gh = FakeGh({issue_url: closed_99_merged_100}) + outcome, pr_url, _reason = pr_outcome_for_issue( + gh, repo="AS215932/network-operations", issue_url=issue_url + ) + assert (outcome, pr_url) == ("success", "u100") + + +def test_null_merged_by_is_pending_not_success() -> None: + issue_url = "https://github.com/AS215932/network-operations/issues/2" + merged_no_actor = json.dumps( + [ + { + "number": 7, + "state": "MERGED", + "url": "u7", + "mergedAt": "2026-07-02T00:00:00Z", + "mergedBy": None, + } + ] + ) + gh = FakeGh({issue_url: merged_no_actor}) + outcome, _url, reason = pr_outcome_for_issue( + gh, repo="AS215932/network-operations", issue_url=issue_url + ) + assert outcome == "pending" + assert "merger unknown" in reason + + +def test_primary_lookup_failure_is_pending_and_reported() -> None: + class RaisingGh(FakeGh): + def run(self, args: list[str]) -> str: + raise RuntimeError("rate limited") + + outcome, _url, reason = pr_outcome_for_issue( + RaisingGh({}), repo="AS215932/network-operations", issue_url="https://x/issues/2" + ) + assert outcome == "pending" + assert "unknown" in reason + + +def test_registry_proposal_counts_only_tier2_successes(tmp_path: Path) -> None: + write_decision_record(_record(2, created_at="2026-07-01T00:00:00Z"), tmp_path) # tier 0 + tier2 = _record(3, created_at="2026-07-02T00:00:00Z") + tier2 = tier2.model_copy(update={"risk_tier": 2, "record_id": "rdr_3b"}) + write_decision_record(tier2, tmp_path) + gh = FakeGh({"issues/2": _merged_pr(7), "issues/3": _merged_pr(8), "Revert": "[]"}) + history = build_capability_history(tmp_path, client=gh, window_days=90, now=NOW) + proposal = history.registry_proposal()["tier0.docs-runbooks-tests"] + # only the tier-2 success feeds the Tier-2 gate; the tier-0 one is reported + # separately, never mixed in + assert proposal["success_count"] == 1 + assert proposal["lower_tier_success_count"] == 1 + assert proposal["failure_count"] == 0 + assert proposal["pending_count"] == 0 diff --git a/tests/test_insight_wiring.py b/tests/test_insight_wiring.py new file mode 100644 index 0000000..abed66d --- /dev/null +++ b/tests/test_insight_wiring.py @@ -0,0 +1,255 @@ +"""Report -> insight mappings and the flag-gated recording path.""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +from agent_core.contracts import InsightDecisionRecord + +from hyrule_engineering_loop import agent_core_trace +from hyrule_engineering_loop.insights import ( + daemon_report_insight, + governor_report_insights, + intake_report_insights, + record_insights, +) + + +def _daemon_report(**overrides) -> dict: + report = {"outcome": "idle", "detail": "", "issue": None, "cost_usd": 0.0, "pr_url": None} + report.update(overrides) + return report + + +def test_daemon_report_insight_action_table() -> None: + published = daemon_report_insight( + _daemon_report( + outcome="published", + pr_url="https://gh/pr/7", + issue={"url": "https://gh/issue/5"}, + ) + ) + assert published is not None + assert (published["action_selected"], published["sampling_class"]) == ("draft", "surfaced") + assert published["budget_context"]["pr_url"] == "https://gh/pr/7" + + triage = daemon_report_insight(_daemon_report(outcome="needs_triage", detail="gates failed")) + assert triage is not None + assert (triage["action_selected"], triage["sampling_class"]) == ("question", "surfaced") + assert "gates failed" in triage["why_now"] + + idle = daemon_report_insight(_daemon_report(outcome="idle")) + assert idle is not None + assert (idle["action_selected"], idle["sampling_class"]) == ( + "stay_silent", + "sampled_quiet_interval", + ) + + for outcome in ("over_budget", "refused_ci", "error"): + withheld = daemon_report_insight(_daemon_report(outcome=outcome)) + assert withheld is not None + assert (withheld["action_selected"], withheld["sampling_class"]) == ( + "stay_silent", + "withheld_logged", + ) + + assert daemon_report_insight(_daemon_report(outcome="locked")) is None + assert daemon_report_insight(_daemon_report(outcome="unknown-outcome")) is None + + +def _decision(issue_id: str, routing: str, **overrides) -> SimpleNamespace: + fields = { + "issue_id": issue_id, + "repo": "AS215932/network-operations", + "issue_number": int(issue_id.rsplit("#", 1)[-1]), + "intent_type": "config_change", + "routing_decision": routing, + "denial_reasons": [], + "policy_rules": ["rule-1"], + "labels_to_add": ["loop:candidate"], + "schema_version": "cdr.v3", + "knowledge_context_pack_id": "ctx_abc", + "knowledge_export_version": "run1:retr1:pol1", + } + fields.update(overrides) + return SimpleNamespace(**fields) + + +def test_governor_report_insights_skips_unchanged_and_maps_routing() -> None: + report = SimpleNamespace( + records=[ + _decision("AS215932/network-operations#10", "allow_candidate"), + _decision("AS215932/network-operations#11", "needs_human", denial_reasons=["tier 3"]), + _decision("AS215932/network-operations#12", "allow_approved"), + ], + skipped=["AS215932/network-operations#12: unchanged decision rdr_1"], + ) + records = governor_report_insights(report, dry_run=False) + assert [r["action_selected"] for r in records] == ["draft", "question"] + assert all(r["sampling_class"] == "surfaced" for r in records) + assert "tier 3" in records[1]["why_now"] + # knowledge context is cited and versioned + kinds = {ref["kind"] for ref in records[0]["evidence_refs"]} + assert "okf_context_pack" in kinds + assert records[0]["tool_versions"] == {"knowledge_export": "run1:retr1:pol1"} + # issue prose is untrusted; the title is structural + assert records[0]["support_facts"][0] == "AS215932/network-operations#10 config_change" + + +def test_governor_report_insights_dry_run_is_withheld() -> None: + report = SimpleNamespace( + records=[_decision("AS215932/network-operations#10", "allow_candidate")], skipped=[] + ) + records = governor_report_insights(report, dry_run=True) + assert records[0]["sampling_class"] == "withheld_logged" + + +def test_intake_report_insights_filed_and_deduped() -> None: + signal = SimpleNamespace( + fingerprint="fp1", + source="ci_failures", + context="CI flaked twice this week.", + action_items=["stabilize test"], + related=["AS215932/network-operations#3"], + ) + report = SimpleNamespace( + filed=[ + { + "title": "Stabilize flaky CI", + "fingerprint": "fp1", + "repo": "AS215932/network-operations", + "source": "ci_failures", + "url": "https://gh/issue/42", + } + ], + deduplicated=[{"title": "Old signal", "fingerprint": "fp2", "existing_issue": 17}], + ) + records = intake_report_insights( + report, [signal], repo="AS215932/network-operations", dry_run=False + ) + assert [r["action_selected"] for r in records] == ["draft", "stay_silent"] + assert records[0]["sampling_class"] == "surfaced" + assert records[1]["sampling_class"] == "withheld_logged" + assert "dedupe" in records[1]["why_now"] + + +def test_all_mapped_records_validate_against_contract() -> None: + report = SimpleNamespace( + records=[_decision("AS215932/network-operations#10", "knowledge_gap")], skipped=[] + ) + rows = governor_report_insights(report, dry_run=False) + rows.append(daemon_report_insight(_daemon_report(outcome="published", pr_url="u"))) + for row in rows: + InsightDecisionRecord.model_validate(row) + + +def test_record_insights_flag_off_writes_nothing(tmp_path, monkeypatch) -> None: + monkeypatch.delenv("HYRULE_ENGINEERING_INSIGHT_RECORDS", raising=False) + count = record_insights( + [daemon_report_insight(_daemon_report(outcome="idle"))], tmp_path + ) + assert count == 0 + assert not (tmp_path / "insights").exists() + + +def test_record_insights_writes_ledger_and_emits(tmp_path, monkeypatch) -> None: + monkeypatch.setenv("HYRULE_ENGINEERING_INSIGHT_RECORDS", "1") + emitted: list[dict] = [] + monkeypatch.setattr( + agent_core_trace, + "emit_insight_decision_envelopes", + lambda records, *, input_event=None: emitted.extend(records) or len(records), + ) + record = daemon_report_insight(_daemon_report(outcome="published", pr_url="u")) + assert record is not None + count = record_insights([record], tmp_path, input_event={"component": "test"}) + assert count == 1 + assert emitted and emitted[0]["insight_id"] == record["insight_id"] + ledger_files = list((tmp_path / "insights").glob("*.jsonl")) + assert len(ledger_files) == 1 + stored = json.loads(ledger_files[0].read_text(encoding="utf-8").strip()) + assert stored["action_selected"] == "draft" + + +def test_emit_insight_decision_envelopes_ships_full_record(tmp_path, monkeypatch) -> None: + trace_path = tmp_path / "trace.jsonl" + monkeypatch.setenv(agent_core_trace.FLAG_ENV, "1") + monkeypatch.setenv(agent_core_trace.PATH_ENV, str(trace_path)) + record = daemon_report_insight(_daemon_report(outcome="idle")) + assert record is not None + delivered = agent_core_trace.emit_insight_decision_envelopes( + [record], input_event={"run_id": "cycle-1"} + ) + assert delivered == 1 + event = json.loads(trace_path.read_text(encoding="utf-8").strip()) + assert event["event_type"] == "loop_decision_envelope" + envelope = event["payload"]["loop_decision_envelope"] + assert envelope["loop"] == "engineering" + # untrusted-text guard fields + run correlation when no explicit trace id + assert event["payload"]["untrusted_loop_text"] is True + assert event["payload"]["model_consumption_allowed"] is False + assert envelope["trace_id"] == "cycle-1" + full = event["payload"]["insight_decision_record"] + assert full["sampling_class"] == "sampled_quiet_interval" + assert full["action_selected"] == "stay_silent" + + +def test_daemon_insight_builds_structural_issue_ref() -> None: + record = daemon_report_insight( + _daemon_report( + outcome="needs_triage", + issue={"repo": "AS215932/network-operations", "number": 42, "title": "t"}, + ) + ) + assert record is not None + ref = "https://github.com/AS215932/network-operations/issues/42" + assert record["evidence_refs"] == [{"kind": "github_issue", "ref": ref}] + + +def test_intake_records_carry_issue_evidence_refs() -> None: + signal = SimpleNamespace( + fingerprint="fp1", source="ci_failures", context="ctx", action_items=[], related=[] + ) + report = SimpleNamespace( + filed=[ + { + "title": "T", + "fingerprint": "fp1", + "repo": "AS215932/network-operations", + "source": "ci_failures", + "url": "https://github.com/AS215932/network-operations/issues/42", + } + ], + deduplicated=[{"title": "Old", "fingerprint": "fp2", "existing_issue": 17}], + ) + filed, deduped = intake_report_insights( + report, [signal], repo="AS215932/network-operations", dry_run=False + ) + assert filed["evidence_refs"][0] == { + "kind": "github_issue", + "ref": "https://github.com/AS215932/network-operations/issues/42", + } + assert deduped["evidence_refs"][0] == { + "kind": "github_issue", + "ref": "https://github.com/AS215932/network-operations/issues/17", + } + + +def test_insight_flag_alone_emits_envelopes(tmp_path, monkeypatch) -> None: + trace_path = tmp_path / "trace.jsonl" + monkeypatch.delenv(agent_core_trace.FLAG_ENV, raising=False) + monkeypatch.setenv("HYRULE_ENGINEERING_INSIGHT_RECORDS", "1") + monkeypatch.setenv(agent_core_trace.PATH_ENV, str(trace_path)) + record = daemon_report_insight(_daemon_report(outcome="idle")) + assert record is not None + delivered = agent_core_trace.emit_insight_decision_envelopes( + [record], input_event={"run_id": "cycle-solo"} + ) + # the insight flag is a complete opt-in: no second master flag required + assert delivered == 1 + event = json.loads(trace_path.read_text(encoding="utf-8").strip()) + assert event["payload"]["insight_decision_record"]["insight_id"] == record["insight_id"] + # and both flags off still emits nothing + monkeypatch.delenv("HYRULE_ENGINEERING_INSIGHT_RECORDS", raising=False) + assert agent_core_trace.emit_insight_decision_envelopes([record], input_event={}) == 0 diff --git a/uv.lock b/uv.lock index 0b45454..574365c 100644 --- a/uv.lock +++ b/uv.lock @@ -12,7 +12,7 @@ resolution-markers = [ [[package]] name = "agent-core" version = "0.7.0" -source = { git = "https://github.com/AS215932/agent-core?tag=v0.7.0#6f79630aeab6f5d9019671fe612a153e286f5321" } +source = { git = "https://github.com/AS215932/agent-core?tag=v0.8.0#4329ec96d689d0dc347b3e420f6c4ac4dcdec945" } dependencies = [ { name = "pydantic" }, ] @@ -365,7 +365,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "agent-core", git = "https://github.com/AS215932/agent-core?tag=v0.7.0" }, + { name = "agent-core", git = "https://github.com/AS215932/agent-core?tag=v0.8.0" }, { name = "langgraph", specifier = ">=0.2.70" }, { name = "mcp", specifier = ">=1.27.0" }, { name = "pydantic", specifier = ">=2.10" },