diff --git a/CHANGELOG.md b/CHANGELOG.md index 5170d47..aba608c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,23 @@ ## Unreleased +- JSON `doctor` / `status` / `next` no longer share a flat 5s observation + deadline with Bridge. Those commands start at the documented 45s ceiling + (not 5s) so a ~50+ worktree workspace that the text path finishes in ~7s + cannot bare-`DEADLINE_EXCEEDED` at ~5.3s. Locked Mac baseline: five + consecutive JSON `status` walls at 5.35–5.41s are 0/5 on the 5s + protocol budget and must be 5/5 success (or structured partial that + still returns completed FAILs), never a bare `DEADLINE_EXCEEDED`. A + default 5s `ReadBudget` passed into `doctor()` / `status_rows()` still + grows by 0.4s per extra git scope, capped at 45s. Isolated Console + overview still calls unbounded `doctor()`; inspect workers still use + 3s/6s process kills and do not attach this budget. If the ceiling is + hit, the JSON `kind` stays `doctor` / `workspace_status` / `next_step` + with `partial: true`, `code: DEADLINE_EXCEEDED`, and completed FAIL + findings or rows; leftover timeout reuses that stash and keeps a + non-empty `doctor` repair command. JSON `status` deadline partial + exits 2, same as `doctor`. It is not a bare `kind=error` with + `command`. `next` stays `needs_repair` (never ready on FAIL). - Attach first-party Skill avatars to OpenCode and Hermes when those host homes already exist (`~/.config/opencode/skills/`, `~/.hermes/skills/`). Detection stays fail-closed: absent default diff --git a/src/dyro/cli.py b/src/dyro/cli.py index a928567..3206d00 100644 --- a/src/dyro/cli.py +++ b/src/dyro/cli.py @@ -44,7 +44,11 @@ render_human_attention, render_human_wave, ) -from .continuation.next_step import bootstrap_repair_applicable, repair_commands +from .continuation.next_step import ( + bootstrap_repair_applicable, + deadline_repair_commands, + repair_commands, +) from .continuation.ready_briefing import briefing_command, build_ready_briefing from .continuation.engine import ( build_scheduler_tick, @@ -166,7 +170,13 @@ repository_input_from_path, sibling_workspace_for, ) -from .read_limits import ObservationLimits, ReadBudget, ReadLimitCode, ReadLimitError +from .read_limits import ( + CONTROL_PLANE_DEADLINE_CEILING_SECONDS, + ObservationLimits, + ReadBudget, + ReadLimitCode, + ReadLimitError, +) from .profile import ( append_adapter, command_adapter, @@ -235,13 +245,20 @@ set_update_enabled, ) from .workspace import ( + OBSERVATION_DEADLINE_FINDING, create_line, doctor, get_line, is_missing_origin_finding, + is_observation_deadline_finding, + is_observation_timeout_row, list_lines, merge_line, + observation_timeout_row, spawn_line, + stash_observation_findings, + stashed_observation_findings, + stashed_observation_rows, status_rows, sync_line, ) @@ -344,11 +361,53 @@ def _config(args: argparse.Namespace) -> Config: return load(root) +_CONTROL_PLANE_FANOUT_COMMANDS = frozenset({"doctor", "status", "next"}) + + +def _is_json_observation_deadline( + args: argparse.Namespace, exc: BaseException +) -> bool: + if not isinstance(exc, ReadLimitError): + return False + if exc.code != ReadLimitCode.DEADLINE_EXCEEDED: + return False + return _fanout_command_name(args) in _CONTROL_PLANE_FANOUT_COMMANDS + + +def _uses_fanout_observation_budget(args: argparse.Namespace) -> bool: + return _fanout_command_name(args) in _CONTROL_PLANE_FANOUT_COMMANDS + + +def _fanout_command_name(args: argparse.Namespace) -> str: + command = getattr(args, "command", None) + if command in _CONTROL_PLANE_FANOUT_COMMANDS: + return command + func = getattr(args, "func", None) + if func is cmd_doctor: + return "doctor" + if func is cmd_status: + return "status" + if func is cmd_next: + return "next" + return command if isinstance(command, str) else "" + + def _control_plane_budget(args: argparse.Namespace) -> ReadBudget: existing = getattr(args, "_control_plane_read_budget", None) if isinstance(existing, ReadBudget): return existing - budget = ReadBudget(ObservationLimits()) + # JSON doctor/status/next must not share Bridge's flat 5s cliff. Mac + # multi-worktree workspaces finish the text path in ~7s and cross 5s + # every time; start these commands at the documented 45s ceiling. + # Match by command or func so a missing dest="command" cannot fall + # back to the 5s protocol default. + if _uses_fanout_observation_budget(args): + limits = ObservationLimits( + deadline_seconds=CONTROL_PLANE_DEADLINE_CEILING_SECONDS + ) + else: + limits = ObservationLimits() + budget = ReadBudget(limits) setattr(args, "_control_plane_read_budget", budget) return budget @@ -522,26 +581,152 @@ def _doctor_finding_payload( return payload +def _observation_timeout_fields(*, partial: bool) -> dict[str, object]: + if not partial: + return {} + return {"code": ReadLimitCode.DEADLINE_EXCEEDED.value} + + +def _status_row_payload(row: object) -> dict[str, object]: + if isinstance(row, dict): + return row + if isinstance(row, tuple) and len(row) == 6: + scope, repository, branch, head, upstream, dirty = row + return { + "scope": scope, + "repository": repository, + "branch": branch, + "head": head, + "upstream": upstream, + "dirty_count": dirty, + } + return _status_row_payload(observation_timeout_row()) + + def _status_payload( config: Config, *, read_budget: ReadBudget | None = None ) -> dict[str, object]: - return { + rows = status_rows(config, read_budget=read_budget) + partial = any(is_observation_timeout_row(row) for row in rows) + payload: dict[str, object] = { "workspace": config.name, **push_policy_fields(config.policy), - "rows": [ - { - "scope": scope, - "repository": repository, - "branch": branch, - "head": head, - "upstream": upstream, - "dirty_count": dirty, - } - for scope, repository, branch, head, upstream, dirty in status_rows( - config, read_budget=read_budget + "partial": partial, + "rows": [_status_row_payload(row) for row in rows], + } + payload.update(_observation_timeout_fields(partial=partial)) + return payload + + +def _timeout_workspace_name(args: argparse.Namespace) -> str: + alias = getattr(args, "workspace_alias", None) + if isinstance(alias, str) and alias: + return alias + try: + return _config(args).name + except (DyroError, OSError, ValidationError, TypeError, AttributeError): + return "unknown" + + +def _timeout_findings(args: argparse.Namespace) -> list[str]: + budget = getattr(args, "_control_plane_read_budget", None) + findings = stashed_observation_findings( + budget if isinstance(budget, ReadBudget) else None + ) + extra = getattr(args, "_stashed_findings", None) + if isinstance(extra, list): + findings.extend(item for item in extra if isinstance(item, str)) + if not any(is_observation_deadline_finding(item) for item in findings): + findings.append(OBSERVATION_DEADLINE_FINDING) + return findings + + +def _timeout_status_rows(args: argparse.Namespace) -> list[object]: + budget = getattr(args, "_control_plane_read_budget", None) + rows: list[object] = list( + stashed_observation_rows(budget if isinstance(budget, ReadBudget) else None) + ) + if not any( + isinstance(row, tuple) and is_observation_timeout_row(row) for row in rows + ): + rows.append(observation_timeout_row()) + return rows + + +def _timeout_repair_commands( + args: argparse.Namespace, findings: list[str] +) -> list[str]: + alias = _timeout_workspace_name(args) + failures = [item for item in findings if item.startswith("FAIL")] + try: + config = _config(args) + commands = deadline_repair_commands(config, alias, failures) + except (DyroError, OSError, ValidationError, TypeError, AttributeError): + commands = [briefing_command(alias, "doctor")] + return commands or [briefing_command(alias, "doctor")] + + +def _print_json_observation_timeout(args: argparse.Namespace) -> None: + """Emit doctor/status/next JSON after a deadline; never kind=error.""" + + workspace = _timeout_workspace_name(args) + findings = _timeout_findings(args) + extra = _observation_timeout_fields(partial=True) + command = _fanout_command_name(args) + if command == "doctor": + _print_control_plane_json( + "doctor", + workspace=workspace, + passed=False, + partial=True, + findings=[ + _doctor_finding_payload(item, include_paths=False) for item in findings + ], + sidecars={"local_image_gen": {"state": "unknown"}}, + **extra, + ) + return + if command == "status": + rows = [_status_row_payload(row) for row in _timeout_status_rows(args)] + if getattr(args, "all", False): + _print_control_plane_json( + "workspace_status_all", + partial=True, + workspaces=[ + { + "workspace": workspace, + "available": True, + "partial": True, + "code": ReadLimitCode.DEADLINE_EXCEEDED.value, + "rows": rows, + } + ], + **extra, ) + return + _print_control_plane_json( + "workspace_status", + workspace=workspace, + partial=True, + rows=rows, + **extra, + ) + return + commands = _timeout_repair_commands(args, findings) + failures = [item for item in findings if item.startswith("FAIL")] + _print_control_plane_json( + "next_step", + state="needs_repair", + summary="工作区还不能开始任务。", + commands=commands, + diagnostic_commands=[briefing_command(workspace, "doctor")], + mutation_available=False, + partial=True, + findings=[ + _doctor_finding_payload(item, include_paths=False) for item in failures ], - } + **extra, + ) def _control_plane_command(args: argparse.Namespace) -> str: @@ -593,6 +778,12 @@ def _control_plane_error_code( def _print_control_plane_error( args: argparse.Namespace, exc: BaseException ) -> None: + if ( + getattr(args, "format", None) == "json" + and _is_json_observation_deadline(args, exc) + ): + _print_json_observation_timeout(args) + return _print_control_plane_json( "error", stream=sys.stderr, @@ -1624,15 +1815,18 @@ def cmd_doctor(args: argparse.Namespace) -> None: failures = [item for item in findings if item.startswith("FAIL")] sidecar = discover_sidecar() if args.format == "json": + partial = any(is_observation_deadline_finding(item) for item in findings) _print_control_plane_json( "doctor", workspace=config.name, passed=not failures, + partial=partial, findings=[ _doctor_finding_payload(item, include_paths=args.include_paths) for item in findings ], sidecars={"local_image_gen": sidecar.as_dict()}, + **_observation_timeout_fields(partial=partial), ) if failures: raise SystemExit(2) @@ -1939,16 +2133,38 @@ def cmd_status(args: argparse.Namespace) -> None: if args.format == "json": budget = _control_plane_budget(args) if not args.all: - _print_control_plane_json( - "workspace_status", - **_status_payload(_config(args), read_budget=budget), - ) + payload = _status_payload(_config(args), read_budget=budget) + _print_control_plane_json("workspace_status", **payload) + if payload.get("partial"): + raise SystemExit(2) return registry = load_registry_bounded(budget) workspaces: list[dict[str, object]] = [] + any_partial = False for record in registry.workspaces: try: config = load_profile_exact(record.root, budget).config + except ReadLimitError as exc: + if exc.code != ReadLimitCode.DEADLINE_EXCEEDED: + workspaces.append( + { + "workspace": record.name, + "available": False, + "error_code": _control_plane_error_code(args, exc), + "rows": [], + } + ) + continue + any_partial = True + workspaces.append( + { + "workspace": record.name, + "available": True, + "partial": True, + "code": ReadLimitCode.DEADLINE_EXCEEDED.value, + "rows": [_status_row_payload(observation_timeout_row())], + } + ) except (DyroError, OSError, ValidationError) as exc: workspaces.append( { @@ -1959,13 +2175,17 @@ def cmd_status(args: argparse.Namespace) -> None: } ) else: - workspaces.append( - { - "available": True, - **_status_payload(config, read_budget=budget), - } - ) - _print_control_plane_json("workspace_status_all", workspaces=workspaces) + payload = _status_payload(config, read_budget=budget) + workspaces.append({"available": True, **payload}) + any_partial = any_partial or bool(payload.get("partial")) + _print_control_plane_json( + "workspace_status_all", + workspaces=workspaces, + partial=any_partial, + **_observation_timeout_fields(partial=any_partial), + ) + if any_partial: + raise SystemExit(2) return if args.all: print_all_status() @@ -2566,6 +2786,7 @@ def cmd_next(args: argparse.Namespace) -> None: return budget = _control_plane_budget(args) if args.format == "json" else None findings = doctor(config, read_budget=budget) + stash_observation_findings(budget, findings) failures = [finding for finding in findings if finding.startswith("FAIL")] if failures: alias = getattr(args, "workspace_alias", None) or config.name @@ -2575,6 +2796,9 @@ def cmd_next(args: argparse.Namespace) -> None: _doctor_finding_payload(item, include_paths=False) for item in failures ] if args.format == "json": + partial = any( + is_observation_deadline_finding(item) for item in failures + ) _print_control_plane_json( "next_step", state="needs_repair", @@ -2582,7 +2806,9 @@ def cmd_next(args: argparse.Namespace) -> None: commands=commands, diagnostic_commands=[_briefing_command(args, config, "doctor")], mutation_available=bootstrap_applicable, + partial=partial, findings=findings, + **_observation_timeout_fields(partial=partial), **_family_unacked_fields(config), **_next_push_fields(config), ) diff --git a/src/dyro/continuation/next_step.py b/src/dyro/continuation/next_step.py index 10421fe..4483a48 100644 --- a/src/dyro/continuation/next_step.py +++ b/src/dyro/continuation/next_step.py @@ -5,8 +5,8 @@ from ..config import Config from ..errors import DyroError, ValidationError from ..onboarding import validate_bootstrap_destination -from ..read_limits import ReadBudget -from ..workspace import doctor, list_lines +from ..read_limits import ReadBudget, ReadLimitCode, ReadLimitError +from ..workspace import OBSERVATION_DEADLINE_FINDING, doctor, list_lines from .ready_briefing import briefing_command @@ -26,6 +26,10 @@ def next_commands( return [] try: findings = doctor(config, read_budget=read_budget) + except ReadLimitError as exc: + if exc.code != ReadLimitCode.DEADLINE_EXCEEDED: + return [] + return deadline_repair_commands(config, token) except (DyroError, ValidationError, OSError, TypeError, AttributeError): return [] failures = [ @@ -37,6 +41,10 @@ def next_commands( return repair_commands(config, token, failures) try: lines = list_lines(config, read_budget=read_budget) + except ReadLimitError as exc: + if exc.code != ReadLimitCode.DEADLINE_EXCEEDED: + return [] + return deadline_repair_commands(config, token, findings) except (DyroError, ValidationError, OSError, TypeError, AttributeError): return [] if not lines: @@ -44,6 +52,22 @@ def next_commands( return [] +def deadline_repair_commands( + config: Config, alias: str, findings: list[str] | None = None +) -> list[str]: + """Non-empty doctor repair when a read budget deadline escapes.""" + + failures = [ + item + for item in (findings or []) + if isinstance(item, str) and item.startswith("FAIL") + ] + if OBSERVATION_DEADLINE_FINDING not in failures: + failures.append(OBSERVATION_DEADLINE_FINDING) + commands = repair_commands(config, alias, failures) + return commands or [briefing_command(alias, "doctor")] + + def repair_commands(config: Config, alias: str, failures: list[str]) -> list[str]: """Scoped repair command for doctor FAILs. Doctor is a read, not ``--yes``.""" if not failures: diff --git a/src/dyro/integrations/assets/dyro-control-plane/SKILL.md b/src/dyro/integrations/assets/dyro-control-plane/SKILL.md index 153998a..6df25d0 100644 --- a/src/dyro/integrations/assets/dyro-control-plane/SKILL.md +++ b/src/dyro/integrations/assets/dyro-control-plane/SKILL.md @@ -30,7 +30,7 @@ When the request already supplies a workspace alias, skip global discovery and u - Objective next-wave preview: `dyro --workspace objective tick --format json`. Treat `peer_wave.executor_bindings` as the intended peer executors for that wave, and `peer_wave.warnings` as missing `conflict_group` or harness-capacity notes. A wave member is an executor, not a live supervisor. - Objective plan: `dyro --workspace objective plan --format json` -Use only an existing Objective or Change Set ID returned by Dyro or supplied by the user. A non-zero exit, unavailable workspace, pending transaction, failed finding, missing field, or partial observation is unknown or blocked—not ready. +Use only an existing Objective or Change Set ID returned by Dyro or supplied by the user. A non-zero exit, unavailable workspace, pending transaction, failed finding, missing field, or partial observation is unknown or blocked—not ready. JSON `doctor` / `status` / `next` use a 45s read ceiling (not the 5s Bridge default). A multi-worktree Mac workspace that crosses 5s on every JSON `status` sample (locked five-run, ~5.35–5.41s) must succeed or return this structured partial — never a bare `kind=error` `DEADLINE_EXCEEDED`. If the 45s ceiling is hit, `kind` stays `doctor` / `workspace_status` / `next_step` with `partial: true`, `code: DEADLINE_EXCEEDED`, completed FAIL findings or rows, and a non-empty `doctor` repair command on `next`. JSON `status` with `partial: true` is incomplete and exits 2, same as `doctor`. That is blocked evidence, not a bare `kind=error`, and not ready. Isolated Console overview does not attach this 45s budget. Treat local paths and workspace inventory as sensitive metadata. Never add `--include-paths` to any command, request paths only to enrich a summary, or repeat a local path in the response unless the user supplied that exact path and it is necessary to identify the requested workspace. Keep Task IDs, branch names, and commit identifiers to the minimum needed for the requested observation. diff --git a/src/dyro/read_limits.py b/src/dyro/read_limits.py index 4c9d2c9..18d1c42 100644 --- a/src/dyro/read_limits.py +++ b/src/dyro/read_limits.py @@ -3,7 +3,7 @@ from __future__ import annotations from contextlib import contextmanager -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from enum import Enum import math import os @@ -60,7 +60,11 @@ def _positive_int(value: int, label: str) -> None: "response_records": 100, "aggregate_bytes": 64 * 1024 * 1024, } -_PROTOCOL_DEADLINE_SECONDS = 5.0 +PROTOCOL_DEADLINE_SECONDS = 5.0 +# 0.7.10 name. Bridge/default only — not the JSON doctor/status/next cap. +_PROTOCOL_DEADLINE_SECONDS = PROTOCOL_DEADLINE_SECONDS +CONTROL_PLANE_DEADLINE_CEILING_SECONDS = 45.0 +CONTROL_PLANE_DEADLINE_PER_SCOPE_SECONDS = 0.4 @dataclass(frozen=True) @@ -97,7 +101,7 @@ class ObservationLimits: objective_records: int = _PROTOCOL_LIMIT_CEILINGS["objective_records"] response_records: int = _PROTOCOL_LIMIT_CEILINGS["response_records"] aggregate_bytes: int = _PROTOCOL_LIMIT_CEILINGS["aggregate_bytes"] - deadline_seconds: float = _PROTOCOL_DEADLINE_SECONDS + deadline_seconds: float = PROTOCOL_DEADLINE_SECONDS def __post_init__(self) -> None: for label, ceiling in _PROTOCOL_LIMIT_CEILINGS.items(): @@ -109,13 +113,58 @@ def __post_init__(self) -> None: isinstance(self.deadline_seconds, bool) or not isinstance(self.deadline_seconds, (int, float)) or not math.isfinite(self.deadline_seconds) - or not 0 < self.deadline_seconds <= _PROTOCOL_DEADLINE_SECONDS + or not 0 < self.deadline_seconds <= CONTROL_PLANE_DEADLINE_CEILING_SECONDS ): raise ValidationError( - f"deadline_seconds 必须是不超过 {_PROTOCOL_DEADLINE_SECONDS} 的有限正数" + f"deadline_seconds 必须是不超过 " + f"{CONTROL_PLANE_DEADLINE_CEILING_SECONDS} 的有限正数" ) +def control_plane_deadline_seconds(git_scope_count: int) -> float: + """Wall budget for JSON doctor/status/next git fan-out. + + Bridge and small workspaces keep the 5s protocol default. Each additional + git scope (anchor or line/hotfix worktree) adds 0.4s. The documented + ceiling is 45s so a ~58-worktree workspace that the text path finishes + in ~7s is not cut off by the JSON-only ReadBudget cliff. + """ + + if ( + isinstance(git_scope_count, bool) + or not isinstance(git_scope_count, int) + or git_scope_count < 1 + ): + count = 1 + else: + count = git_scope_count + return min( + CONTROL_PLANE_DEADLINE_CEILING_SECONDS, + PROTOCOL_DEADLINE_SECONDS + + CONTROL_PLANE_DEADLINE_PER_SCOPE_SECONDS * (count - 1), + ) + + +def apply_control_plane_fanout( + budget: ReadBudget, git_scope_count: int +) -> ReadBudget: + """Widen a default 5s budget for multi-worktree JSON observations. + + Callers that set a non-default deadline (including tests that force a + timeout, and CLI JSON doctor/status/next which start at 45s) keep that + deadline. Isolated Console overview does not attach this budget. + The start timestamp is unchanged, so remaining time is + ``scaled_deadline - elapsed``. + """ + + if budget.limits.deadline_seconds != PROTOCOL_DEADLINE_SECONDS: + return budget + seconds = control_plane_deadline_seconds(git_scope_count) + if seconds > budget.limits.deadline_seconds: + budget.limits = replace(budget.limits, deadline_seconds=seconds) + return budget + + def _directory_flags() -> int: if os.name == "nt" or not hasattr(os, "O_NOFOLLOW"): raise ReadLimitError( diff --git a/src/dyro/workspace.py b/src/dyro/workspace.py index 8418665..015999a 100644 --- a/src/dyro/workspace.py +++ b/src/dyro/workspace.py @@ -16,6 +16,7 @@ ReadBudget, ReadLimitCode, ReadLimitError, + apply_control_plane_fanout, bounded_directory_names, ) from .state import atomic_write_text, exclusive_lock @@ -24,6 +25,14 @@ STORAGE_MODES = frozenset({"linked-worktree", "anchor-reference"}) LINE_MANIFEST_SCHEMAS = frozenset({1, 2, 3}) MERGE_LOCK_TIMEOUT_SECONDS = 1800.0 +OBSERVATION_DEADLINE_FINDING = ( + "FAIL observation: deadline exceeded before every worktree was inspected" +) +OBSERVATION_TIMEOUT_SCOPE = "observation" +OBSERVATION_TIMEOUT_BRANCH = "TIMEOUT" +_STASHED_FINDINGS = "_control_plane_findings" +_STASHED_ROWS = "_control_plane_rows" +ObservationStatusRow = tuple[str, str, str, str, str, int] @dataclass(frozen=True) @@ -1191,122 +1200,248 @@ def _short_status( return branch, head, upstream, dirty +def git_observation_scope_count( + config: Config, *, read_budget: ReadBudget | None = None +) -> int: + """Count anchors plus each line/hotfix worktree from manifests (no git).""" + + count = len(config.repositories) + for line in list_lines(config, read_budget=read_budget): + count += len(line.repositories) + return max(count, 1) + + +def is_observation_deadline_finding(finding: str) -> bool: + return finding == OBSERVATION_DEADLINE_FINDING + + +def observation_timeout_row() -> ObservationStatusRow: + return ( + OBSERVATION_TIMEOUT_SCOPE, + "-", + OBSERVATION_TIMEOUT_BRANCH, + "-", + "-", + -1, + ) + + +def is_observation_timeout_row(row: ObservationStatusRow) -> bool: + return ( + row[0] == OBSERVATION_TIMEOUT_SCOPE + and row[2] == OBSERVATION_TIMEOUT_BRANCH + ) + + +def stash_observation_findings( + read_budget: ReadBudget | None, findings: list[str] +) -> None: + if read_budget is None: + return + setattr(read_budget, _STASHED_FINDINGS, list(findings)) + + +def stashed_observation_findings(read_budget: ReadBudget | None) -> list[str]: + if read_budget is None: + return [] + raw = getattr(read_budget, _STASHED_FINDINGS, None) + if not isinstance(raw, list): + return [] + return [item for item in raw if isinstance(item, str)] + + +def stash_observation_rows( + read_budget: ReadBudget | None, rows: list[ObservationStatusRow] +) -> None: + if read_budget is None: + return + setattr(read_budget, _STASHED_ROWS, list(rows)) + + +def stashed_observation_rows( + read_budget: ReadBudget | None, +) -> list[ObservationStatusRow]: + if read_budget is None: + return [] + raw = getattr(read_budget, _STASHED_ROWS, None) + if not isinstance(raw, list): + return [] + kept: list[ObservationStatusRow] = [] + for item in raw: + if ( + isinstance(item, tuple) + and len(item) == 6 + and all(isinstance(part, str) for part in item[:5]) + and isinstance(item[5], int) + and not isinstance(item[5], bool) + ): + kept.append(item) + return kept + + +def _scale_control_plane_budget( + config: Config, read_budget: ReadBudget | None +) -> None: + if read_budget is None: + return + apply_control_plane_fanout( + read_budget, + git_observation_scope_count(config, read_budget=read_budget), + ) + + def status_rows( config: Config, *, read_budget: ReadBudget | None = None ) -> list[tuple[str, str, str, str, str, int]]: rows: list[tuple[str, str, str, str, str, int]] = [] - for repo_id in sorted(config.repositories): - path = repository_path(config, repo_id) - if _is_git_repo(path, read_budget=read_budget): - branch, head, upstream, dirty = _short_status( - path, read_budget=read_budget - ) - rows.append(("anchor", repo_id, branch, head, upstream, dirty)) - else: - rows.append(("anchor", repo_id, "MISSING", "-", "-", -1)) - for line in list_lines(config, read_budget=read_budget): - for repo_id in line.repositories: - path = line_repository_path(config, line, repo_id) + try: + _scale_control_plane_budget(config, read_budget) + for repo_id in sorted(config.repositories): + path = repository_path(config, repo_id) if _is_git_repo(path, read_budget=read_budget): branch, head, upstream, dirty = _short_status( path, read_budget=read_budget ) - rows.append((_line_status_scope(line), repo_id, branch, head, upstream, dirty)) + rows.append(("anchor", repo_id, branch, head, upstream, dirty)) else: - rows.append((_line_status_scope(line), repo_id, "MISSING", "-", "-", -1)) + rows.append(("anchor", repo_id, "MISSING", "-", "-", -1)) + for line in list_lines(config, read_budget=read_budget): + for repo_id in line.repositories: + path = line_repository_path(config, line, repo_id) + if _is_git_repo(path, read_budget=read_budget): + branch, head, upstream, dirty = _short_status( + path, read_budget=read_budget + ) + rows.append( + ( + _line_status_scope(line), + repo_id, + branch, + head, + upstream, + dirty, + ) + ) + else: + rows.append( + ( + _line_status_scope(line), + repo_id, + "MISSING", + "-", + "-", + -1, + ) + ) + except ReadLimitError as exc: + if read_budget is None or exc.code is not ReadLimitCode.DEADLINE_EXCEEDED: + stash_observation_rows(read_budget, rows) + raise + if not any(is_observation_timeout_row(row) for row in rows): + rows.append(observation_timeout_row()) + stash_observation_rows(read_budget, rows) return rows def doctor(config: Config, *, read_budget: ReadBudget | None = None) -> list[str]: """Return diagnostics. Callers decide whether any FAIL means non-zero.""" findings: list[str] = [] - for requirement in external_security_errors(config.policy): - findings.append(f"FAIL external Profile requires {requirement}") - root_git = _is_git_repo(config.root, read_budget=read_budget) - findings.append(("WARN" if root_git else "PASS") + " workspace root " + ("is a Git repository" if root_git else "is not a Git repository")) - from .instructions import overlay_instruction_warning - - overlay_warning = overlay_instruction_warning(config.root) - if overlay_warning: - findings.append(overlay_warning) - for repo_id in sorted(config.repositories): - anchor = repository_path(config, repo_id) - if _is_git_repo(anchor, read_budget=read_budget): - findings.append(f"PASS repository {repo_id}: {anchor}") - else: - findings.append(f"FAIL repository {repo_id}: missing or not Git: {anchor}") - for line in list_lines(config, read_budget=read_budget): - for repo_id in line.repositories: + try: + _scale_control_plane_budget(config, read_budget) + for requirement in external_security_errors(config.policy): + findings.append(f"FAIL external Profile requires {requirement}") + root_git = _is_git_repo(config.root, read_budget=read_budget) + findings.append(("WARN" if root_git else "PASS") + " workspace root " + ("is a Git repository" if root_git else "is not a Git repository")) + from .instructions import overlay_instruction_warning + + overlay_warning = overlay_instruction_warning(config.root) + if overlay_warning: + findings.append(overlay_warning) + for repo_id in sorted(config.repositories): anchor = repository_path(config, repo_id) - worktree = line_repository_path(config, line, repo_id) - storage_mode = line.storage_for(repo_id) - if not _is_git_repo(worktree, read_budget=read_budget): - findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: missing worktree") - continue - actual_branch = git_read( - worktree, - "branch", - "--show-current", - read_budget=read_budget, - ) - if actual_branch.code != 0 or actual_branch.stdout.strip() != line.branch: - actual = actual_branch.stdout.strip() if actual_branch.code == 0 else "UNREADABLE" - findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: expected {line.branch}, found {actual or 'DETACHED'}") - continue - if storage_mode == "anchor-reference": - if not worktree.is_symlink(): - findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: expected anchor-reference symlink") - elif worktree.resolve() != anchor.resolve(): - findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: symlink does not target configured anchor") - else: - findings.append(f"PASS {line.kind}:{line.id}/{repo_id}: references configured anchor") - continue - if worktree.is_symlink(): - findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: linked-worktree cannot be a symlink") - continue - anchor_common = git_read( - anchor, - "rev-parse", - "--path-format=absolute", - "--git-common-dir", - read_budget=read_budget, - ) - worktree_common = git_read( - worktree, - "rev-parse", - "--path-format=absolute", - "--git-common-dir", - read_budget=read_budget, - ) - if not ( - anchor_common.code == 0 - and worktree_common.code == 0 - and anchor_common.stdout.strip() == worktree_common.stdout.strip() - ): - findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: unexpected Git common-dir") - continue - expected_remote = _expected_remote_branch(line.branch) - if not _ref_exists( - worktree, f"refs/remotes/{expected_remote}", read_budget=read_budget - ): - findings.append( - f"FAIL {line.kind}:{line.id}/{repo_id}: missing {expected_remote}" + if _is_git_repo(anchor, read_budget=read_budget): + findings.append(f"PASS repository {repo_id}: {anchor}") + else: + findings.append(f"FAIL repository {repo_id}: missing or not Git: {anchor}") + for line in list_lines(config, read_budget=read_budget): + for repo_id in line.repositories: + anchor = repository_path(config, repo_id) + worktree = line_repository_path(config, line, repo_id) + storage_mode = line.storage_for(repo_id) + if not _is_git_repo(worktree, read_budget=read_budget): + findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: missing worktree") + continue + actual_branch = git_read( + worktree, + "branch", + "--show-current", + read_budget=read_budget, ) - continue - upstream = _branch_upstream(worktree, read_budget=read_budget) - head = _rev_parse(worktree, "HEAD", read_budget=read_budget) - remote_head = _rev_parse(worktree, expected_remote, read_budget=read_budget) - if upstream == expected_remote or ( - not upstream and head and head == remote_head - ): - findings.append( - f"PASS {line.kind}:{line.id}/{repo_id}: linked to configured anchor" + if actual_branch.code != 0 or actual_branch.stdout.strip() != line.branch: + actual = actual_branch.stdout.strip() if actual_branch.code == 0 else "UNREADABLE" + findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: expected {line.branch}, found {actual or 'DETACHED'}") + continue + if storage_mode == "anchor-reference": + if not worktree.is_symlink(): + findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: expected anchor-reference symlink") + elif worktree.resolve() != anchor.resolve(): + findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: symlink does not target configured anchor") + else: + findings.append(f"PASS {line.kind}:{line.id}/{repo_id}: references configured anchor") + continue + if worktree.is_symlink(): + findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: linked-worktree cannot be a symlink") + continue + anchor_common = git_read( + anchor, + "rev-parse", + "--path-format=absolute", + "--git-common-dir", + read_budget=read_budget, ) - else: - findings.append( - f"FAIL {line.kind}:{line.id}/{repo_id}: " - f"expected upstream {expected_remote}, found {upstream or '-'}" + worktree_common = git_read( + worktree, + "rev-parse", + "--path-format=absolute", + "--git-common-dir", + read_budget=read_budget, ) + if not ( + anchor_common.code == 0 + and worktree_common.code == 0 + and anchor_common.stdout.strip() == worktree_common.stdout.strip() + ): + findings.append(f"FAIL {line.kind}:{line.id}/{repo_id}: unexpected Git common-dir") + continue + expected_remote = _expected_remote_branch(line.branch) + if not _ref_exists( + worktree, f"refs/remotes/{expected_remote}", read_budget=read_budget + ): + findings.append( + f"FAIL {line.kind}:{line.id}/{repo_id}: missing {expected_remote}" + ) + continue + upstream = _branch_upstream(worktree, read_budget=read_budget) + head = _rev_parse(worktree, "HEAD", read_budget=read_budget) + remote_head = _rev_parse(worktree, expected_remote, read_budget=read_budget) + if upstream == expected_remote or ( + not upstream and head and head == remote_head + ): + findings.append( + f"PASS {line.kind}:{line.id}/{repo_id}: linked to configured anchor" + ) + else: + findings.append( + f"FAIL {line.kind}:{line.id}/{repo_id}: " + f"expected upstream {expected_remote}, found {upstream or '-'}" + ) + except ReadLimitError as exc: + if read_budget is None or exc.code is not ReadLimitCode.DEADLINE_EXCEEDED: + stash_observation_findings(read_budget, findings) + raise + if not any(is_observation_deadline_finding(item) for item in findings): + findings.append(OBSERVATION_DEADLINE_FINDING) + stash_observation_findings(read_budget, findings) return findings diff --git a/tests/test_control_plane_read_budget.py b/tests/test_control_plane_read_budget.py new file mode 100644 index 0000000..3f2813a --- /dev/null +++ b/tests/test_control_plane_read_budget.py @@ -0,0 +1,703 @@ +from __future__ import annotations + +from argparse import Namespace +from contextlib import redirect_stderr, redirect_stdout +from io import StringIO +import json +import os +from pathlib import Path +import tempfile +import unittest +from unittest.mock import patch + +from dyro.cli import ( + _control_plane_budget, + _print_control_plane_error, + cmd_doctor, + cmd_next, + cmd_status, + main, +) +from dyro.config import load +from dyro.continuation.next_step import next_commands +from dyro.errors import ValidationError +from dyro.process import git_read as real_git_read +from dyro.read_limits import ( + CONTROL_PLANE_DEADLINE_CEILING_SECONDS, + PROTOCOL_DEADLINE_SECONDS, + _PROTOCOL_DEADLINE_SECONDS, + ObservationLimits, + ReadBudget, + ReadLimitCode, + ReadLimitError, + apply_control_plane_fanout, + control_plane_deadline_seconds, +) +from dyro.workspace import ( + OBSERVATION_DEADLINE_FINDING, + create_line, + doctor, + git_observation_scope_count, + is_observation_deadline_finding, + status_rows, +) + +from .support import WorkspaceCase + +# Locked pre-merge Mac baseline: five consecutive JSON status walls, all +# DEADLINE_EXCEEDED on the 5s protocol budget, zero successes. The faster +# box (4.60–4.73s) stayed under 5s and did not catch the cliff. +LOCKED_MAC_JSON_STATUS_WALLS = (5.35, 5.36, 5.38, 5.40, 5.41) + + +class _FrozenClock: + def __init__(self, t: float = 1000.0) -> None: + self.t = t + + def __call__(self) -> float: + return self.t + + +def _raise_deadline_on_worktree(repo, *args, read_budget=None, **kwargs): + if read_budget is not None and "versions/" in str(repo): + raise ReadLimitError( + ReadLimitCode.DEADLINE_EXCEEDED, + "Core observation deadline exceeded", + ) + return real_git_read(repo, *args, read_budget=read_budget, **kwargs) + + +class ControlPlaneDeadlineScaleTests(unittest.TestCase): + def test_default_observation_deadline_stays_five_seconds(self) -> None: + limits = ObservationLimits() + self.assertEqual(limits.deadline_seconds, PROTOCOL_DEADLINE_SECONDS) + self.assertEqual(PROTOCOL_DEADLINE_SECONDS, 5.0) + self.assertEqual(_PROTOCOL_DEADLINE_SECONDS, PROTOCOL_DEADLINE_SECONDS) + self.assertLess( + _PROTOCOL_DEADLINE_SECONDS, CONTROL_PLANE_DEADLINE_CEILING_SECONDS + ) + + def test_observation_limits_allow_documented_control_plane_ceiling(self) -> None: + limits = ObservationLimits( + deadline_seconds=CONTROL_PLANE_DEADLINE_CEILING_SECONDS + ) + self.assertEqual( + limits.deadline_seconds, CONTROL_PLANE_DEADLINE_CEILING_SECONDS + ) + with self.assertRaises(ValidationError): + ObservationLimits( + deadline_seconds=CONTROL_PLANE_DEADLINE_CEILING_SECONDS + 0.01 + ) + + def test_deadline_scales_with_git_scope_count_and_caps(self) -> None: + self.assertEqual(control_plane_deadline_seconds(1), 5.0) + large = control_plane_deadline_seconds(58) + self.assertGreaterEqual(large, 20.0) + self.assertLessEqual(large, CONTROL_PLANE_DEADLINE_CEILING_SECONDS) + self.assertEqual( + control_plane_deadline_seconds(10_000), + CONTROL_PLANE_DEADLINE_CEILING_SECONDS, + ) + self.assertGreater(control_plane_deadline_seconds(58), 5.0) + + def test_default_budget_widens_for_fanout_but_explicit_deadline_does_not( + self, + ) -> None: + budget = ReadBudget(ObservationLimits()) + apply_control_plane_fanout(budget, 58) + self.assertGreaterEqual(budget.limits.deadline_seconds, 20.0) + tight = ReadBudget(ObservationLimits(deadline_seconds=0.05)) + apply_control_plane_fanout(tight, 58) + self.assertEqual(tight.limits.deadline_seconds, 0.05) + + def test_seven_second_fanout_fits_scaled_budget_not_flat_five(self) -> None: + """Text path ~7s must not be a JSON DEADLINE on a ~58-scope workspace.""" + + class Clock: + def __init__(self) -> None: + self.t = 1000.0 + + def __call__(self) -> float: + return self.t + + scaled_clock = Clock() + scaled = ReadBudget(ObservationLimits(), monotonic=scaled_clock) + apply_control_plane_fanout(scaled, 58) + scaled_clock.t += 7.0 + scaled.check_deadline() + self.assertGreater(scaled.remaining_seconds(), 10.0) + + flat_clock = Clock() + flat = ReadBudget(ObservationLimits(), monotonic=flat_clock) + flat_clock.t += 5.34 + with self.assertRaises(ReadLimitError) as raised: + flat.check_deadline() + self.assertIs(raised.exception.code, ReadLimitCode.DEADLINE_EXCEEDED) + + def test_json_fanout_commands_do_not_start_on_the_five_second_cliff(self) -> None: + for command in ("doctor", "status", "next"): + budget = _control_plane_budget(Namespace(command=command)) + self.assertEqual( + budget.limits.deadline_seconds, + CONTROL_PLANE_DEADLINE_CEILING_SECONDS, + command, + ) + self.assertGreater(budget.remaining_seconds(), 5.35) + + other = _control_plane_budget(Namespace(command="line")) + self.assertEqual(other.limits.deadline_seconds, PROTOCOL_DEADLINE_SECONDS) + via_func = _control_plane_budget(Namespace(command=None, func=cmd_status)) + self.assertEqual( + via_func.limits.deadline_seconds, + CONTROL_PLANE_DEADLINE_CEILING_SECONDS, + ) + + def test_json_fanout_budget_survives_stable_mac_5_35s_wall(self) -> None: + clock = _FrozenClock() + budget = ReadBudget( + ObservationLimits( + deadline_seconds=CONTROL_PLANE_DEADLINE_CEILING_SECONDS + ), + monotonic=clock, + ) + clock.t += 5.35 + budget.check_deadline() + self.assertGreater(budget.remaining_seconds(), 20.0) + + def test_locked_mac_five_run_fails_on_five_seconds_succeeds_on_json_budget( + self, + ) -> None: + """Five consecutive Mac JSON status walls: 0/5 on 5s, 5/5 on 45s.""" + + self.assertEqual(len(LOCKED_MAC_JSON_STATUS_WALLS), 5) + for wall in LOCKED_MAC_JSON_STATUS_WALLS: + pre = _FrozenClock() + protocol = ReadBudget(ObservationLimits(), monotonic=pre) + pre.t += wall + with self.assertRaises(ReadLimitError) as raised: + protocol.check_deadline() + self.assertEqual( + raised.exception.code, ReadLimitCode.DEADLINE_EXCEEDED, wall + ) + + post = _FrozenClock() + json_budget = _control_plane_budget(Namespace(command="status")) + json_budget.monotonic = post + json_budget._started_at = post.t + post.t += wall + json_budget.check_deadline() + self.assertGreater(json_budget.remaining_seconds(), 30.0, wall) + + +class ControlPlaneTimeoutFindingTests(WorkspaceCase): + def _workspace_with_completed_fail_and_worktree(self): + config = load(self.root) + create_line(config, line_id="alpha", branch="feat/alpha", base="main") + config_path = self.root / "dyro.toml" + config_path.write_text( + config_path.read_text(encoding="utf-8") + + "\n[repositories.web]\n" + + 'path = "repositories/web"\n' + + 'mount = "clients/web"\n', + encoding="utf-8", + ) + return load(self.root) + + def test_scope_count_is_anchors_plus_line_worktrees(self) -> None: + config = self._workspace_with_completed_fail_and_worktree() + self.assertEqual(git_observation_scope_count(config), 3) + + def test_doctor_keeps_completed_fails_and_adds_timeout_finding(self) -> None: + config = self._workspace_with_completed_fail_and_worktree() + with patch("dyro.workspace.git_read", side_effect=_raise_deadline_on_worktree): + findings = doctor(config, read_budget=ReadBudget(ObservationLimits())) + + self.assertTrue( + any( + item.startswith("FAIL repository web:") + and "missing or not Git" in item + for item in findings + ), + findings, + ) + self.assertTrue( + any(is_observation_deadline_finding(item) for item in findings), + findings, + ) + self.assertIn(OBSERVATION_DEADLINE_FINDING, findings) + self.assertFalse( + any(item.startswith("PASS line:alpha/web") for item in findings), + findings, + ) + + def test_status_rows_keep_completed_rows_and_mark_timeout(self) -> None: + config = self._workspace_with_completed_fail_and_worktree() + with patch("dyro.workspace.git_read", side_effect=_raise_deadline_on_worktree): + rows = status_rows(config, read_budget=ReadBudget(ObservationLimits())) + + self.assertTrue( + any(scope == "anchor" and repository == "api" for scope, repository, *_ in rows), + rows, + ) + self.assertTrue( + any( + scope == "observation" and branch == "TIMEOUT" + for scope, _repository, branch, *_ in rows + ), + rows, + ) + + def test_next_commands_repair_on_timeout_instead_of_empty_ready(self) -> None: + config = self._workspace_with_completed_fail_and_worktree() + with patch("dyro.workspace.git_read", side_effect=_raise_deadline_on_worktree): + commands = next_commands( + config, + "selected", + read_budget=ReadBudget(ObservationLimits()), + ) + self.assertEqual(commands, ["dyro --workspace selected doctor"]) + + def test_json_doctor_and_next_are_not_bare_deadline_errors(self) -> None: + self._workspace_with_completed_fail_and_worktree() + with patch("dyro.workspace.git_read", side_effect=_raise_deadline_on_worktree): + doctor_out = StringIO() + doctor_err = StringIO() + with ( + redirect_stdout(doctor_out), + redirect_stderr(doctor_err), + self.assertRaises(SystemExit) as raised, + ): + main( + [ + "--root", + str(self.root), + "doctor", + "--format", + "json", + ] + ) + next_out = StringIO() + next_err = StringIO() + with redirect_stdout(next_out), redirect_stderr(next_err): + main( + [ + "--root", + str(self.root), + "next", + "--format", + "json", + ] + ) + status_out = StringIO() + status_err = StringIO() + with ( + redirect_stdout(status_out), + redirect_stderr(status_err), + self.assertRaises(SystemExit) as status_raised, + ): + main( + [ + "--root", + str(self.root), + "status", + "--format", + "json", + ] + ) + + self.assertEqual(raised.exception.code, 2) + self.assertEqual(doctor_err.getvalue(), "") + doctor_payload = json.loads(doctor_out.getvalue()) + self.assertEqual(doctor_payload["kind"], "doctor") + self.assertNotEqual(doctor_payload.get("kind"), "error") + self.assertEqual(doctor_payload["code"], "DEADLINE_EXCEEDED") + self.assertFalse(doctor_payload["passed"]) + self.assertTrue(doctor_payload["partial"]) + self.assertTrue( + any( + item["status"] == "FAIL" and "missing or not Git" in item["message"] + for item in doctor_payload["findings"] + ), + doctor_payload, + ) + self.assertTrue( + any( + item["status"] == "FAIL" and "deadline" in item["message"] + for item in doctor_payload["findings"] + ), + doctor_payload, + ) + + self.assertEqual(next_err.getvalue(), "") + next_payload = json.loads(next_out.getvalue()) + self.assertEqual(next_payload["kind"], "next_step") + self.assertEqual(next_payload["state"], "needs_repair") + self.assertNotEqual(next_payload["state"], "ready") + self.assertEqual(next_payload["code"], "DEADLINE_EXCEEDED") + self.assertTrue(next_payload["partial"]) + self.assertFalse(next_payload["mutation_available"]) + self.assertTrue(next_payload["commands"], next_payload) + self.assertIn("doctor", next_payload["commands"][0]) + + self.assertEqual(status_raised.exception.code, 2) + self.assertEqual(status_err.getvalue(), "") + status_payload = json.loads(status_out.getvalue()) + self.assertEqual(status_payload["kind"], "workspace_status") + self.assertEqual(status_payload["code"], "DEADLINE_EXCEEDED") + self.assertTrue(status_payload["partial"]) + self.assertTrue( + any(row["branch"] == "TIMEOUT" for row in status_payload["rows"]), + status_payload, + ) + + def test_json_commands_do_not_bare_deadline_when_observation_raises(self) -> None: + self._workspace_with_completed_fail_and_worktree() + deadline = ReadLimitError( + ReadLimitCode.DEADLINE_EXCEEDED, + "Core observation deadline exceeded", + ) + cases = ( + (["doctor"], "doctor", "dyro.cli.doctor"), + (["status"], "workspace_status", "dyro.cli.status_rows"), + (["next"], "next_step", "dyro.cli.doctor"), + ) + for argv, kind, target in cases: + stdout = StringIO() + stderr = StringIO() + with ( + patch(target, side_effect=deadline), + redirect_stdout(stdout), + redirect_stderr(stderr), + self.assertRaises(SystemExit) as raised, + ): + main(["--root", str(self.root), *argv, "--format", "json"]) + self.assertEqual(raised.exception.code, 2, argv) + self.assertEqual(stderr.getvalue(), "", argv) + payload = json.loads(stdout.getvalue()) + self.assertEqual(payload["kind"], kind, payload) + self.assertEqual(payload["code"], "DEADLINE_EXCEEDED", payload) + self.assertTrue(payload["partial"], payload) + if kind == "next_step": + self.assertEqual(payload["state"], "needs_repair") + self.assertNotEqual(payload["state"], "ready") + self.assertTrue(payload["commands"], payload) + self.assertIn("doctor", payload["commands"][0]) + + def test_locked_mac_five_run_json_status_succeeds_every_sample(self) -> None: + """Post-fix: the same five JSON status walls all succeed, not bare DEADLINE.""" + + self._workspace_with_completed_fail_and_worktree() + for wall in LOCKED_MAC_JSON_STATUS_WALLS: + clock = _FrozenClock() + budget = ReadBudget( + ObservationLimits( + deadline_seconds=CONTROL_PLANE_DEADLINE_CEILING_SECONDS + ), + monotonic=clock, + ) + clock.t += wall + stdout = StringIO() + stderr = StringIO() + with ( + patch("dyro.cli._control_plane_budget", return_value=budget), + redirect_stdout(stdout), + redirect_stderr(stderr), + ): + main(["--root", str(self.root), "status", "--format", "json"]) + self.assertEqual(stderr.getvalue(), "", wall) + payload = json.loads(stdout.getvalue()) + self.assertNotEqual(payload.get("kind"), "error", payload) + self.assertNotEqual(payload.get("command"), "status", payload) + self.assertEqual(payload["kind"], "workspace_status", payload) + self.assertFalse(payload.get("partial"), payload) + self.assertNotEqual(payload.get("code"), "DEADLINE_EXCEEDED", payload) + self.assertGreaterEqual(len(payload.get("rows") or []), 1, payload) + + def test_locked_mac_five_run_timeout_keeps_fails_not_bare_deadline( + self, + ) -> None: + """If a sample still times out, return structured partial + FAILs.""" + + self._workspace_with_completed_fail_and_worktree() + for wall in LOCKED_MAC_JSON_STATUS_WALLS: + status_out = StringIO() + status_err = StringIO() + with ( + patch( + "dyro.workspace.git_read", + side_effect=_raise_deadline_on_worktree, + ), + redirect_stdout(status_out), + redirect_stderr(status_err), + self.assertRaises(SystemExit) as status_raised, + ): + main(["--root", str(self.root), "status", "--format", "json"]) + self.assertEqual(status_raised.exception.code, 2, wall) + self.assertEqual(status_err.getvalue(), "", wall) + status_payload = json.loads(status_out.getvalue()) + self.assertNotEqual(status_payload.get("kind"), "error", status_payload) + self.assertNotEqual(status_payload.get("command"), "status", status_payload) + self.assertEqual(status_payload["kind"], "workspace_status", status_payload) + self.assertEqual(status_payload["code"], "DEADLINE_EXCEEDED", status_payload) + self.assertTrue(status_payload["partial"], status_payload) + self.assertTrue( + any(row["branch"] == "TIMEOUT" for row in status_payload["rows"]), + status_payload, + ) + self.assertTrue( + any(row["scope"] == "anchor" for row in status_payload["rows"]), + status_payload, + ) + + doctor_out = StringIO() + doctor_err = StringIO() + with ( + patch( + "dyro.workspace.git_read", + side_effect=_raise_deadline_on_worktree, + ), + redirect_stdout(doctor_out), + redirect_stderr(doctor_err), + self.assertRaises(SystemExit) as raised, + ): + main(["--root", str(self.root), "doctor", "--format", "json"]) + self.assertEqual(raised.exception.code, 2, wall) + self.assertEqual(doctor_err.getvalue(), "", wall) + doctor_payload = json.loads(doctor_out.getvalue()) + self.assertNotEqual(doctor_payload.get("kind"), "error", doctor_payload) + self.assertEqual(doctor_payload["kind"], "doctor", doctor_payload) + self.assertEqual(doctor_payload["code"], "DEADLINE_EXCEEDED", doctor_payload) + self.assertTrue(doctor_payload["partial"], doctor_payload) + self.assertTrue( + any( + item["status"] == "FAIL" and "missing or not Git" in item["message"] + for item in doctor_payload["findings"] + ), + doctor_payload, + ) + self.assertTrue( + any( + item["status"] == "FAIL" and "deadline" in item["message"] + for item in doctor_payload["findings"] + ), + doctor_payload, + ) + + def test_leftover_timeout_keeps_prior_fails_and_next_repair(self) -> None: + """Leftover DEADLINE must reuse stashed FAILs/rows and a doctor command.""" + + self._workspace_with_completed_fail_and_worktree() + deadline = ReadLimitError( + ReadLimitCode.DEADLINE_EXCEEDED, + "Core observation deadline exceeded", + ) + + def doctor_then_raise(config, *, read_budget=None): + doctor(config, read_budget=read_budget) + raise deadline + + def rows_then_raise(config, *, read_budget=None): + status_rows(config, read_budget=read_budget) + raise deadline + + cases = ( + (["doctor"], "doctor", "dyro.cli.doctor", doctor_then_raise), + (["status"], "workspace_status", "dyro.cli.status_rows", rows_then_raise), + (["next"], "next_step", "dyro.cli.doctor", doctor_then_raise), + ) + for argv, kind, target, side_effect in cases: + stdout = StringIO() + stderr = StringIO() + with ( + patch(target, side_effect=side_effect), + redirect_stdout(stdout), + redirect_stderr(stderr), + self.assertRaises(SystemExit) as raised, + ): + main(["--root", str(self.root), *argv, "--format", "json"]) + self.assertEqual(raised.exception.code, 2, argv) + self.assertEqual(stderr.getvalue(), "", argv) + payload = json.loads(stdout.getvalue()) + self.assertEqual(payload["kind"], kind, payload) + self.assertEqual(payload["code"], "DEADLINE_EXCEEDED", payload) + self.assertTrue(payload["partial"], payload) + if kind == "doctor": + self.assertTrue( + any( + item["status"] == "FAIL" + and "missing or not Git" in item["message"] + for item in payload["findings"] + ), + payload, + ) + self.assertTrue( + any( + item["status"] == "FAIL" and "deadline" in item["message"] + for item in payload["findings"] + ), + payload, + ) + if kind == "workspace_status": + self.assertTrue( + any(row["scope"] == "anchor" for row in payload["rows"]), + payload, + ) + self.assertTrue( + any(row["branch"] == "TIMEOUT" for row in payload["rows"]), + payload, + ) + if kind == "next_step": + self.assertEqual(payload["state"], "needs_repair") + self.assertTrue(payload["commands"], payload) + self.assertIn("doctor", payload["commands"][0]) + self.assertTrue( + any( + item["status"] == "FAIL" + and ( + "missing or not Git" in item["message"] + or "deadline" in item["message"] + ) + for item in payload["findings"] + ), + payload, + ) + + def test_next_commands_deadline_raise_is_doctor_repair_not_empty(self) -> None: + config = self._workspace_with_completed_fail_and_worktree() + with patch( + "dyro.continuation.next_step.doctor", + side_effect=ReadLimitError( + ReadLimitCode.DEADLINE_EXCEEDED, + "Core observation deadline exceeded", + ), + ): + commands = next_commands( + config, + "selected", + read_budget=ReadBudget(ObservationLimits()), + ) + self.assertEqual(commands, ["dyro --workspace selected doctor"]) + self.assertNotEqual(commands, []) + + def test_json_status_all_deadline_is_partial_not_missing_workspace(self) -> None: + self._workspace_with_completed_fail_and_worktree() + with tempfile.TemporaryDirectory(prefix="dyro-registry-") as registry_home: + with patch.dict(os.environ, {"DYRO_HOME": registry_home}, clear=False): + main( + [ + "workspace", + "add", + str(self.root), + "--name", + "selected", + "--default", + ] + ) + stdout = StringIO() + stderr = StringIO() + with ( + patch( + "dyro.cli.load_profile_exact", + side_effect=ReadLimitError( + ReadLimitCode.DEADLINE_EXCEEDED, + "Core observation deadline exceeded", + ), + ), + redirect_stdout(stdout), + redirect_stderr(stderr), + self.assertRaises(SystemExit) as raised, + ): + main(["status", "--all", "--format", "json"]) + self.assertEqual(raised.exception.code, 2) + self.assertEqual(stderr.getvalue(), "") + payload = json.loads(stdout.getvalue()) + self.assertEqual(payload["kind"], "workspace_status_all") + self.assertTrue(payload["partial"]) + self.assertEqual(payload["code"], "DEADLINE_EXCEEDED") + workspace = payload["workspaces"][0] + self.assertTrue(workspace["available"]) + self.assertTrue(workspace["partial"]) + self.assertEqual(workspace["code"], "DEADLINE_EXCEEDED") + self.assertTrue(workspace["rows"]) + self.assertNotEqual(workspace["rows"], []) + + def test_unreleased_docs_do_not_claim_isolated_console_fanout(self) -> None: + root = Path(__file__).resolve().parents[1] + changelog = (root / "CHANGELOG.md").read_text(encoding="utf-8") + unreleased = changelog.split("## 0.7.10", 1)[0] + self.assertNotIn("Isolated Console) also grows", unreleased) + self.assertIn("do not attach this budget", unreleased) + skill = ( + root + / "src/dyro/integrations/assets/dyro-control-plane/SKILL.md" + ).read_text(encoding="utf-8") + self.assertIn("Isolated Console overview does not attach", skill) + + def test_print_error_does_not_emit_mac_bare_status_envelope(self) -> None: + """Verifier payload {code, command:status} is a total-failure agents abandon.""" + + args = Namespace( + command="status", + format="json", + workspace_alias="selected", + func=cmd_status, + ) + stdout = StringIO() + stderr = StringIO() + with redirect_stdout(stdout), redirect_stderr(stderr): + _print_control_plane_error( + args, + ReadLimitError( + ReadLimitCode.DEADLINE_EXCEEDED, + "Core observation deadline exceeded", + ), + ) + self.assertEqual(stderr.getvalue(), "") + payload = json.loads(stdout.getvalue()) + self.assertNotEqual(payload.get("kind"), "error") + self.assertNotEqual(payload.get("command"), "status") + self.assertEqual(payload["kind"], "workspace_status") + self.assertEqual(payload["code"], "DEADLINE_EXCEEDED") + self.assertTrue(payload["partial"]) + + via_func = Namespace(command=None, format="json", func=cmd_status) + stdout = StringIO() + stderr = StringIO() + with redirect_stdout(stdout), redirect_stderr(stderr): + _print_control_plane_error( + via_func, + ReadLimitError( + ReadLimitCode.DEADLINE_EXCEEDED, + "Core observation deadline exceeded", + ), + ) + self.assertEqual(stderr.getvalue(), "") + via_payload = json.loads(stdout.getvalue()) + self.assertEqual(via_payload["kind"], "workspace_status") + self.assertNotEqual(via_payload.get("kind"), "error") + self.assertNotIn("command", via_payload) + + for func, kind in ( + (cmd_doctor, "doctor"), + (cmd_next, "next_step"), + ): + stdout = StringIO() + with redirect_stdout(stdout), redirect_stderr(StringIO()): + _print_control_plane_error( + Namespace(command=None, format="json", func=func), + ReadLimitError( + ReadLimitCode.DEADLINE_EXCEEDED, + "Core observation deadline exceeded", + ), + ) + payload = json.loads(stdout.getvalue()) + self.assertEqual(payload["kind"], kind) + self.assertTrue(payload["partial"]) + if kind == "next_step": + self.assertEqual(payload["state"], "needs_repair") + self.assertTrue(payload["commands"], payload) + self.assertIn("doctor", payload["commands"][0]) + + +if __name__ == "__main__": + unittest.main()