diff --git a/README.md b/README.md index e2e18f7..b185900 100644 --- a/README.md +++ b/README.md @@ -64,6 +64,7 @@ Pick the row that matches what you want to do. | run a recorded cycle from the command line | [`run_cycle.py`](docs/workflows.md#runtime-driver-run_cyclepy) | via its stages | | close a cycle whose rereview ran outside the runtime | [`manual_rereview.py`](docs/review-cycle.md#manual-rereview) | publishes one PR comment | | list or revoke stored sandbox permissions | [`consent_control.py`](docs/routing.md#sensitive-sandbox-grants) | — | +| inspect quota records or attest a legacy handoff | [`quota_state.py`](docs/routing.md#operator-quota-recovery) | — | | review a diff or branch without a pull request | [`cc-code-review`](docs/skills.md#cc-code-review) | — | | audit a change for security | [`cc-security-review`](docs/skills.md#cc-security-review) | — | | prove a change actually works | [`cc-verify`](docs/skills.md#cc-verify) | — | diff --git a/docs/instrumentation.md b/docs/instrumentation.md index c8cb285..1036421 100644 --- a/docs/instrumentation.md +++ b/docs/instrumentation.md @@ -766,11 +766,13 @@ resolution only when it names a merged PR in the resolving snapshot. Schema 23 adds `quota_exhaustions`, keyed by repository, executor, and requested model, with exhaustion and expiry timestamps, a closed reason token, and whether -the executor reported a reset time. Production `--continue` reads active rows; -expired rows are deleted when read, and a successful production dispatch clears -its matching row. The additive table does not rewrite existing telemetry, and a -code rollback leaves unused rows. The `quota_resume_reason` cycle payload field -records when a continuation routes around a persisted exhaustion. +the executor reported a reset time. All production routes read active rows. +Schema 26 keeps expired records and adds independent `quota_history` and +`quota_handoffs` audit tables; success ends availability blocking without deleting +provenance. Unknown dated resets use `reset_time_ambiguous` and no recovery time. +The `quota_resume_reason` cycle payload field records a route around persisted +exhaustion. The journal records redacted messages and scoped handoff fingerprints, +executors and operator actions. See [quota telemetry](telemetry.md#quota-exhaustion-windows-schema-23). Schema 24 adds the stop reasons `publication_violation` and `publication_unverified` of the [publication audit](role-workspace-policy.md#publication-audit). diff --git a/docs/routing.md b/docs/routing.md index 94fb951..e87421f 100644 --- a/docs/routing.md +++ b/docs/routing.md @@ -111,8 +111,20 @@ Resolve keeps the `cheap_coder` / `deep_coder` profile choice. With `code_cycle. - **Probe once per run.** An executor that is only *installed* isn't dispatchable. Codex and Claude can only prove `authenticated`: remaining quota and session validity are invisible until something is spent. Production uses the `attempt` readiness policy to dispatch from `authenticated`, and records that it did (`dispatched_from`). - **Fallback (production only).** If the primary is unavailable or not eligible for the role's workspace policy, the profile's `fallback` is used, and the decision says so. - **Sandbox recovery.** An explicit, complete agent-reported restriction can use the same reroute machinery. See [Recovering sandbox restrictions](#recovering-sandbox-restrictions) for its additional conditions. -- **One reroute per stage.** A `capacity`, `transport`, `timeout`, or `quota` failure can use that profile's distinct, available fallback once, sharing the quota retry budget. The previous process group must have exited; asynchronous launches and missing exit proof stop. A `not_started` attempt can reroute directly. A `started` or `unknown` attempt that could write or publish first reconciles the workspace, local HEAD, remote refs, PR state, comments and reviews, and issue comments. Dirty or changed state, a publication, or unavailable evidence stops with a closed reason token. A publication is what the cycle's own code-host account (`gh api user`) created or edited: pull requests (state, head, title, body), comments and reviews. Other accounts' comments and newly opened pull requests, reactions and timestamps do not stop the reroute; an unreadable account is unavailable evidence. A pull request carries no editor identity, so any edit to an existing in-scope pull request stops conservatively, whoever opened it. A review's read-only workspace does not exempt its publication boundary. Other error classes and human friction never reroute. Both attempts and the parent's `error_code` and `start_state` are recorded. The fallback applies only to this stage; later stages keep the original executor availability. Recovery from published evidence remains a separate operation. -- **Quota across `--continue`.** A quota failure is saved in `telemetry.sqlite` by repository, executor, and requested model. If the dirty-workspace guard stops the same-dispatch fallback, the next `--continue --cwd ` uses the profile fallback while that exact model's record is active, carrying the partial work as its starting point. Other models and executors remain eligible. A reported time of day expires at its next occurrence in the local timezone; with no parseable reset (or usable local timezone), the record lasts 30 minutes. A successful dispatch on that target clears it early. If recording fails, the cycle continues without persistent evidence. When no configured target remains, the stop names the earliest reset. The stop report directs the operator to wait for reset or continue on the fallback. Calibration never reads or writes these records and never substitutes an arm. +- **One reroute per stage.** A `capacity`, `transport`, `timeout`, or `quota` failure can use that profile's distinct, available fallback once, sharing the quota retry budget. The previous process group must have exited; asynchronous launches and missing exit proof stop. A `not_started` attempt can reroute directly. A `started` or `unknown` attempt that could write or publish first reconciles the workspace, local HEAD, remote refs, PR state, comments and reviews, and issue comments. Dirty or changed state, a publication, or unavailable evidence stops with a closed reason token. A publication is what the cycle's own code-host account (`gh api user`) created or edited: pull requests (state, head, title, body), comments and reviews. Other accounts' comments and newly opened pull requests, reactions and timestamps do not stop the reroute; an unreadable account is unavailable evidence. A pull request carries no editor identity, so any edit to an existing in-scope pull request stops conservatively, whoever opened it. A review's read-only workspace does not exempt its publication boundary. Other error classes and human friction never reroute. Both attempts and the parent's `error_code` and `start_state` are recorded. The fallback applies only to this stage; later stages keep the original executor availability except for recorded quota exhaustion, which they continue to honour. Recovery from published evidence remains a separate operation. +- **Quota across resumes.** Every production route skips an active quota record for the exact repository, executor and requested model, including `--continue` and `--from`. The route reports the substitution and reset or operator-retry requirement. Calibration never reads or writes these records and never substitutes an arm. Complete dates (including ordinal days), years, clocks and explicit UTC offsets or IANA zones (including parenthesized zones) are preserved. Without an explicit zone the runtime's local zone applies, retaining its date-specific DST rules; when no IANA zone is available, including on Windows, the OS converts the local wall time for that date. A repeated hour uses its later interpretation. Uninterpretable dates, weekday-only dates, relative durations and ambiguous zone abbreviations stay unavailable until an explicit operator retry. Missing reset information alone uses a 30-minute minimum. Expiry is retry eligibility, not proof of recovery. Original provider text is redacted into the journal and stop report; telemetry holds closed reason tokens and structured timestamps only. +- **Dirty `--from` resumes.** A dirty resume needs an exact quota worktree provenance record bound to the repository, work item, stage, PR, HEAD, original executor/model, destination executor and complete SHA-256 fingerprint of staged, unstaged and nonignored untracked changes. The runtime records ownership only after a finished quota-exhausted process began on a clean tree (or matched previous provenance), left HEAD unchanged and changed no remote or publication boundary. The fallback must inspect those edits before writing. A missing record, altered diff, wrong scope or HEAD stops without changing the checkout. Ordinary dirty executor changes remain refused; `implement --continue` retains its existing behavior. Provenance survives quota expiry, allowing a matching resume on the original executor after expiry; a new exhaustion records another ownership checkpoint. A successful resume consumes the checkpoint, retaining its audit history. This does not weaken the PR-head guard or any sandbox policy. + +### Operator quota recovery + +Use `python3 scripts/quota_state.py --database --repo --task `: + +- `list` shows exhaustion record IDs, timestamps, ambiguity tokens and retry authorization history, without provider prose. +- `retry --executor --record ` explicitly authorizes retry of that exact current record, records the authenticated `gh api user` identity and journal timestamp, and preserves exhaustion history. It does not claim quota recovered. Another quota failure creates a new record. Superseded or already authorized records are refused. +- `inspect --cwd --pr --stage --original --model --destination --head ` previews the exact fingerprint and scope for a legacy worktree. +- `attest` takes the same scope plus `--confirm-fingerprint `. The operator must inspect and explicitly confirm that exact fingerprint. Only a legacy quota record predating automatic provenance history is eligible. A new unattested worktree cannot use this path. The attestation records the authenticated operator and timestamp as `operator_attested`, never `executor_verified`; it is bound to the exact scope and diff, can be created once for that scope, and is claimed atomically before one dispatch. It is never functional verification. + +For interrupted implementation use `run_cycle.py ... --continue --cwd `. For an existing PR use `run_cycle.py ... --from resolve --pr --cwd ` (or the interrupted review stage), without `--continue`. These are alternatives; the CLI rejects combining them. A legacy worktree is never automatically attested, and no WIP commit is needed. Native POSIX dispatches run in their own process group and stop remaining children before returning exit proof, including on timeout. The group first receives `SIGTERM`, so a sandbox can unmount and remove what it set up, then `SIGKILL` after a bounded grace period (`PROCESS_GROUP_GRACE_SECONDS`, 10 seconds); output pipes a stray descendant keeps open are waited on for at most `OUTPUT_DRAIN_SECONDS`, so cleanup never blocks the cycle indefinitely. Windows currently lacks group exit proof and therefore stops rather than automatically rerouting. Reconciliation reads GitHub through `gh`; unsupported publication boundaries fail closed. Snapshots are transient and never enter telemetry. diff --git a/docs/telemetry.md b/docs/telemetry.md index dbf5cf1..e970c24 100644 --- a/docs/telemetry.md +++ b/docs/telemetry.md @@ -499,13 +499,24 @@ comparison of the two policies on a controlled fake CI. ### Quota exhaustion windows (schema 23) The `quota_exhaustions` table persists a quota reset window per repository, -executor, and requested model so a later production `--continue` can skip that -exact target and use the configured fallback. A reported time of day expires at -its next occurrence in the local timezone; an absent or unparseable reset uses a -30-minute window. Reading active rows deletes expired entries. A successful -production dispatch clears only its matching row. Calibration does not read, -write, or clear these records. The table is additive; reverting the code leaves -its rows unused. +executor, and requested model so every production route, including `--from`, +can skip that exact target. Schema 26 preserves expired rows: reading availability +never deletes ownership or audit evidence. An empty projection `expires_at` +means `reset_time_ambiguous`, with no invented reset timestamp. History uses SQL +NULL for that unknown time. Only missing reset information uses a 30-minute +window. Success ends matching eligibility without deleting history. Calibration +does not read, write or clear these records. + +Schema 26 adds `quota_history` with an immutable exhaustion ID, executor/model, +exhaustion time, nullable reset, closed reason and optional explicit `retry_at`. +Each repeated exhaustion appends a new record. It also adds `quota_handoffs`, +whose repository/work-item/stage/PR/HEAD scope, original and destination executors, +original model, SHA-256 diff fingerprint, provenance, operator and creation time +outlive quota availability. `claimed_at` makes operator attestations single-use; +`used_at` marks successful resumes without removing audit rows. File contents, +diff text and provider prose never enter these tables. The journal holds the +redacted original message and operator actions. See [operator quota recovery](routing.md#operator-quota-recovery). +The additive tables leave existing rows readable; rollback leaves them unused. ### Work-item outcomes (schema 14) diff --git a/scripts/cycle.py b/scripts/cycle.py index dd0df35..021494a 100644 --- a/scripts/cycle.py +++ b/scripts/cycle.py @@ -38,7 +38,7 @@ from datetime import datetime, timezone from typing import Any, cast -from cycle_journal import Journal +from cycle_journal import Journal, redact from execution_plan import RoleContract, role_contracts from executors import ( Availability, @@ -59,6 +59,7 @@ observed_sandbox_blocker, denied_result, ) from reroute import effect_snapshot, reconcile +from quota_state import Handoffs from jev_shadow import JevShadow, telemetry_fields as shadow_fields from router import ( RoutingDecision, @@ -286,11 +287,11 @@ def __init__( self.availability = dict(availability) self.quota_exhaustions: list[dict] = [] self.exhausted_until: dict[tuple[str, str], str] = {} - if continue_work and mode is RoutingMode.PRODUCTION: + if mode is RoutingMode.PRODUCTION: try: self.quota_exhaustions = telemetry.active_quota_exhaustions(repo_id) self.exhausted_until = { - (row["executor"], row["model"]): row["expires_at"] + (row["executor"], row["model"]): row["expires_at"] or "operator retry required" for row in self.quota_exhaustions } except (OSError, sqlite3.Error, TelemetryError, ValueError): @@ -527,14 +528,18 @@ def select(candidates): if (self.mode is RoutingMode.PRODUCTION and decision.used_fallback and decision.target is not None and any(f"{row['executor']}/{row['model']} quota exhausted until " - f"{row['expires_at']}" in decision.reasons + f"{row['expires_at'] or 'operator retry required'}" in decision.reasons for row in self.quota_exhaustions)): signals["quota_resume_reason"] = "recorded_quota_exhaustion" return decision decision = select(eligible) if role == "resolve" and not decision.blocked and decision.target is not None: - if target_override is not None: + if (target_override is not None + and (target_override.executor, target_override.model) in self.exhausted_until): + decision = replace(decision, reasons=decision.reasons + + ("previous resolve target has recorded quota exhaustion; keep the available route",)) + elif target_override is not None: if (target_override.executor not in eligible or not self.availability.get( target_override.executor, Availability.UNKNOWN).dispatchable): @@ -632,6 +637,22 @@ def select(candidates): if attempt == 0: first = (decision, signals) primary_decision = decision + handoff = None + if (attempt == 0 and role == self.started_from + and self.started_from != "implement" + and self.mode is RoutingMode.PRODUCTION + and dispatch_kwargs.get("cwd") and not decision.blocked): + try: + from executors import _git_output, _workspace_fingerprint + cwd = dispatch_kwargs["cwd"] + if _git_output(cwd, "status", "--porcelain", "--untracked-files=all"): + handoff = Handoffs(self.telemetry.path).match( + repo=self.repo_id, task=self.task_id, stage=role, + pr=self.change_request_id, workspace=_workspace_fingerprint(cwd), + destination=decision.target) + except (OSError, ValueError, sqlite3.Error) as error: + decision = replace(decision, blocked=True, target=None, + reasons=decision.reasons + (str(error),)) if decision.blocked: outcome.decision = decision self._journal("failure", role=role, component="router", @@ -711,6 +732,24 @@ def snapshot(): before = (snapshot() if (writes or publishes) and self.mode is not RoutingMode.CALIBRATION else None) + if handoff is not None: + try: + if before is None or before["workspace"]["fingerprint"] != handoff["fingerprint"]: + raise ValueError("quota handoff diff changed before dispatch") + Handoffs(self.telemetry.path).claim(handoff) + except (OSError, ValueError, sqlite3.Error) as error: + raise CycleError(str(error)) from error + task += ( + "\n\nQuota worktree resume: inspect the existing staged, unstaged and untracked " + "changes before editing. Preserve correct partial work; do not reset or overwrite it. " + f"Original executor: {handoff['original_executor']}; continuing executor: " + f"{decision.target.executor}; diff SHA-256: {handoff['fingerprint']}; " + f"provenance: {handoff['provenance']}. This is not functional verification.") + self._journal("stage_result", role=role, component="quota", status="handoff", + correlation={"original_executor": handoff["original_executor"], + "destination_executor": decision.target.executor, + "fingerprint": handoff["fingerprint"], + "provenance": handoff["provenance"]}) def observe_progress(**activity): nonlocal tool_activity @@ -910,6 +949,32 @@ def reroute_gate(): self.repo_id, decision.target.executor, decision.target.model, result.detail, ) + self.exhausted_until[(decision.target.executor, decision.target.model)] = ( + exhaustion_expiry or "operator retry required") + self._journal("failure", role=role, component="quota", status="exhausted", + detail=result.detail, + correlation={"expires_at": exhaustion_expiry, + "ambiguity": "reset_time_ambiguous" if exhaustion_expiry is None else None}) + after = snapshot() if self.change_request_id else None + # Only a finished executor with a clean initial tree, + # or a validated prior ownership record, can own edits. + if (result.process_exited and not result.asynchronous + and not result.needs_human_action and fallback is not None + and before is not None and after is not None + and (not before["dirty"] or handoff is not None) + and after["dirty"] + and before["workspace"]["head"] == after["workspace"]["head"] + and reconcile({**before, "dirty": False}, + {**after, "dirty": False, "workspace": before["workspace"]})[0] == "reroute"): + record_id = Handoffs(self.telemetry.path).record( + repo=self.repo_id, task=self.task_id, stage=role, + pr=self.change_request_id, workspace=after["workspace"], + original=decision.target, destination=fallback) + self._journal("stage_result", role=role, component="quota", status="provenance_recorded", + correlation={"record_id": record_id, + "original_executor": decision.target.executor, + "destination_executor": fallback.executor, + "fingerprint": after["workspace"]["fingerprint"]}) except (OSError, sqlite3.Error, TelemetryError, ValueError): # Telemetry failure must not block the writing stage. pass @@ -939,23 +1004,25 @@ def reroute_gate(): "reroute_action": action, "reroute_reason": reason, } else: - detail = f"{result.detail}; reroute stopped: {reason}" + detail = f"{redact(result.detail) if result.failure_code == 'quota' else result.detail}; reroute stopped: {reason}" if (result.failure_code == "quota" and self.mode is RoutingMode.PRODUCTION): if exhaustion_expiry and reset_reported: detail += (f"; quota exhaustion recorded until {exhaustion_expiry}; " - "wait for the reset or run --continue to use the configured fallback") + "wait for the reset or " + self._quota_resume_command(role)) elif exhaustion_expiry: detail += ("; quota reset is unknown; retry after 30 minutes; " f"temporary quota window recorded until {exhaustion_expiry}; " - "run --continue to use the configured fallback") + + self._quota_resume_command(role)) else: - detail += ("; quota reset is unknown; retry after 30 minutes or " - "run --continue to use the configured fallback") + detail += ("; quota reset is ambiguous; executor unavailable until explicit operator retry; " + + self._quota_resume_command(role)) result = replace(result, detail=detail) elif (result.outcome is DispatchOutcome.SUCCEEDED and self.mode is RoutingMode.PRODUCTION): try: + if handoff is not None: + Handoffs(self.telemetry.path).complete(handoff) self.telemetry.clear_quota_exhaustion( self.repo_id, decision.target.executor, decision.target.model, ) @@ -1067,6 +1134,12 @@ def reroute_gate(): self._observe_shadow(role, *first) return outcome + def _quota_resume_command(self, role: str) -> str: + if self.started_from == "implement" and role == "implement": + return "run --continue to use the configured fallback" + return (f"resume with --from {role} --pr {self.change_request_id} " + "to use the configured fallback") + def _journal(self, kind: str, **fields) -> None: if self.journal is not None: fields.setdefault("stage_seq", self.stage_seq) diff --git a/scripts/cycle_dispatch.py b/scripts/cycle_dispatch.py index 10263e5..78b5990 100644 --- a/scripts/cycle_dispatch.py +++ b/scripts/cycle_dispatch.py @@ -85,6 +85,9 @@ def run(self: Any, role: str, instruction: str = "", *, stage_kwargs = {**stage_kwargs, "command_policy": command_policy} if change_request_id: self.journal.change_request_id = change_request_id + # The PR can first become known after implementation. Keep the + # quota checkpoint and its effect snapshot on that same scope. + self.recorder.change_request_id = change_request_id # A resumed or re-dispatched stage sees what earlier runs did (#306). redispatched = any(stage.role == role for stage in self.report.stages) journal_context = (self.journal.summary() diff --git a/scripts/quota_state.py b/scripts/quota_state.py new file mode 100644 index 0000000..f4aace5 --- /dev/null +++ b/scripts/quota_state.py @@ -0,0 +1,303 @@ +"""Quota reset interpretation and durable, scoped worktree handoff evidence. + +No provider prose or file contents are stored here. The journal owns redacted +messages. Availability and provenance deliberately have separate lifecycles. +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sqlite3 +import subprocess +import uuid +from contextlib import closing +from datetime import datetime, timedelta, timezone +from pathlib import Path +from zoneinfo import ZoneInfo + +QUOTA_SCHEMA = """ +CREATE TABLE IF NOT EXISTS quota_history ( + id TEXT PRIMARY KEY, repo_id TEXT NOT NULL, executor TEXT NOT NULL, + model TEXT NOT NULL, exhausted_at TEXT NOT NULL, expires_at TEXT, + reason TEXT NOT NULL, retry_at TEXT +); +CREATE TABLE IF NOT EXISTS quota_handoffs ( + id TEXT PRIMARY KEY, repo_id TEXT NOT NULL, task_id TEXT NOT NULL, + stage TEXT NOT NULL, pr TEXT NOT NULL, head TEXT NOT NULL, + original_executor TEXT NOT NULL, original_model TEXT NOT NULL, + destination_executor TEXT NOT NULL, fingerprint TEXT NOT NULL, + provenance TEXT NOT NULL, operator TEXT, created_at TEXT NOT NULL, + claimed_at TEXT, used_at TEXT +); +CREATE UNIQUE INDEX IF NOT EXISTS quota_operator_attestation_once + ON quota_handoffs(repo_id,task_id,stage,pr) WHERE provenance='operator_attested'; +""" + + +def local_zone(): + """Use the runtime's local zone, retaining DST rules for a future date.""" + name = os.environ.get("TZ", "").removeprefix(":") + try: + if name: + return ZoneInfo(name) + with Path("/etc/localtime").open("rb") as handle: + return ZoneInfo.from_file(handle) + except (OSError, ValueError, KeyError): + # Windows normally has neither TZ nor /etc/localtime. Keep local wall + # times naive there so astimezone() applies OS rules for their date. + return None + + +def quota_reset(detail: str | None, now: datetime | None = None): + """UTC reset, reported flag, closed uncertainty token. + + An explicit invalid date/zone is never reduced to just its time of day. + Missing reset information alone gets the minimum 30 minute window. + """ + current = now or datetime.now(timezone.utc) + text = re.sub(r"(?<=\d)(st|nd|rd|th)\b", "", detail or "", flags=re.I) + clock = re.search(r"(? 14 or mins > 59 or (hours == 14 and mins): + raise ValueError("invalid zone") + minutes = (hours * 60 + mins) * (-1 if offset[0] == "-" else 1) + zone = timezone(timedelta(minutes=minutes)) + elif suffix: + # Only an actual IANA zone is unambiguous; CST, IST, etc. aren't. + if not re.fullmatch(r"[A-Za-z0-9_+.-]+(?:/[A-Za-z0-9_+.-]+)+", suffix): + raise ValueError("ambiguous timezone") + zone = ZoneInfo(suffix) + # With no IANA zone, preserve naive wall time until final conversion; + # today's fixed UTC offset is not valid for every future date. + local = current.astimezone(zone).replace(tzinfo=zone) + prefix = text[:clock.start()].strip() + date_match = re.search(r"\b([A-Za-z]{3,9})\s+(\d{1,2}),?\s+(\d{4})\b", prefix) + iso_match = re.search(r"\b(\d{4}-\d{2}-\d{2})[T ]?$", prefix) + if date_match: + date_text = date_match[0].replace(",", "") + try: + date = datetime.strptime(date_text, "%b %d %Y") + except ValueError: + date = datetime.strptime(date_text, "%B %d %Y") + candidate = date.replace(hour=hour, minute=minute, second=second, tzinfo=zone) + elif iso_match: + candidate = datetime.fromisoformat(iso_match[1]).replace( + hour=hour, minute=minute, second=second, tzinfo=zone) + else: + if re.search(r"\d|\b(?:Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec|" + r"Mon|Tue|Wed|Thu|Fri|Sat|Sun)\w*\b|\b(?:today|tomorrow|next)\b", prefix, re.I): + raise ValueError("unresolved date") + candidate = local.replace(hour=hour, minute=minute, second=second, microsecond=0) + # During a repeated local hour choose the later defensible instant. + reset = max(candidate.replace(fold=fold).astimezone(timezone.utc) + for fold in (0, 1)) + if not date_match and not iso_match and reset <= current: + candidate += timedelta(days=1) + reset = max(candidate.replace(fold=fold).astimezone(timezone.utc) + for fold in (0, 1)) + return reset.isoformat(), True, "quota" + except (ValueError, OverflowError, KeyError, OSError): + return None, True, "reset_time_ambiguous" + + +class Handoffs: + def __init__(self, path): + self.path = path + + def connect(self): + connection = sqlite3.connect(self.path) + connection.row_factory = sqlite3.Row + return connection + + def record(self, *, repo, task, stage, pr, workspace, original, destination, + provenance="executor_verified", operator=None): + if not pr or not workspace or not re.fullmatch(r"[0-9a-f]{40,64}", workspace.get("head", "")): + raise ValueError("handoff needs a PR and full HEAD") + if not re.fullmatch(r"[0-9a-f]{64}", workspace.get("fingerprint", "")): + raise ValueError("handoff needs a complete diff fingerprint") + record_id = f"handoff-{uuid.uuid4().hex}" + with closing(self.connect()) as connection, connection: + if provenance == "operator_attested" and connection.execute( + "SELECT 1 FROM quota_handoffs WHERE repo_id=? AND task_id=? AND stage=? AND pr=?", + (repo, task, stage, str(pr))).fetchone(): + raise ValueError("legacy attestation already recorded for this scope") + connection.execute( + "INSERT INTO quota_handoffs (id,repo_id,task_id,stage,pr,head," + "original_executor,original_model,destination_executor,fingerprint," + "provenance,operator,created_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", + (record_id, repo, task, stage, str(pr), workspace["head"], + original.executor, original.model, destination.executor, + workspace["fingerprint"], provenance, operator, + datetime.now(timezone.utc).isoformat())) + return record_id + + def match(self, *, repo, task, stage, pr, workspace, destination): + with closing(self.connect()) as connection: + rows = connection.execute( + "SELECT * FROM quota_handoffs WHERE repo_id=? AND task_id=? AND stage=? " + "AND pr=? AND used_at IS NULL ORDER BY rowid DESC", + (repo, task, stage, str(pr))).fetchall() + if not rows: + raise ValueError("dirty resume has no quota worktree provenance") + row = dict(rows[0]) + if row["claimed_at"] and row["provenance"] == "operator_attested": + raise ValueError("operator attestation already used") + if workspace is None or row["head"] != workspace["head"]: + raise ValueError("quota handoff HEAD mismatch") + if row["fingerprint"] != workspace["fingerprint"]: + raise ValueError("quota handoff diff fingerprint mismatch") + if destination.executor not in {row["original_executor"], row["destination_executor"]}: + raise ValueError("quota handoff destination executor mismatch") + if destination.executor == row["original_executor"] and destination.model != row["original_model"]: + raise ValueError("quota handoff original model mismatch") + return row + + def claim(self, row): + if row["provenance"] != "operator_attested": + return + with closing(self.connect()) as connection, connection: + updated = connection.execute( + "UPDATE quota_handoffs SET claimed_at=? WHERE id=? AND claimed_at IS NULL AND used_at IS NULL", + (datetime.now(timezone.utc).isoformat(), row["id"])) + if updated.rowcount != 1: + raise ValueError("operator attestation already used") + + def complete(self, row): + with closing(self.connect()) as connection, connection: + connection.execute( + "UPDATE quota_handoffs SET used_at=? WHERE repo_id=? AND task_id=? AND stage=? AND pr=? AND used_at IS NULL", + (datetime.now(timezone.utc).isoformat(), row["repo_id"], row["task_id"], row["stage"], row["pr"])) + + +def main(argv=None): + from telemetry import Telemetry + from executors import _workspace_fingerprint + from router import Target + from cycle_journal import record_event, _identifier, _change_request_identifier + + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--database", required=True) + parser.add_argument("--repo", required=True) + parser.add_argument("--task", required=True) + commands = parser.add_subparsers(dest="command", required=True) + commands.add_parser("list", help="list quota record IDs and reset uncertainty") + retry = commands.add_parser("retry", help="authorize one target to be dispatched again") + retry.add_argument("--record", required=True) + retry.add_argument("--executor", required=True, choices=("codex", "claude", "orca")) + for name in ("inspect", "attest"): + command = commands.add_parser(name) + command.add_argument("--cwd", required=True) + command.add_argument("--pr", required=True) + command.add_argument("--stage", required=True, choices=("review", "resolve", "rereview")) + command.add_argument("--original", required=True, choices=("codex", "claude", "orca")) + command.add_argument("--model", required=True) + command.add_argument("--destination", required=True, choices=("codex", "claude", "orca")) + command.add_argument("--head", required=True) + if name == "attest": + command.add_argument("--confirm-fingerprint", required=True) + args = parser.parse_args(argv) + # Validate operator supplied references before writing durable state. + try: + _identifier(args.repo) + _identifier(args.task) + if args.command not in {"retry", "list"}: + args.pr = _change_request_identifier(args.pr, args.repo) + except ValueError as error: + parser.exit(2, f"quota recovery refused: {error}\n") + store = Telemetry(args.database) + ledger = Handoffs(store.path) + try: + if args.command == "list": + with closing(ledger.connect()) as connection: + rows = connection.execute( + "SELECT * FROM quota_history WHERE repo_id=? ORDER BY rowid DESC", (args.repo,)).fetchall() + print(json.dumps([dict(row) for row in rows])) + return 0 + workspace = None + if args.command != "retry": + store._checked("model_requested", args.model) + workspace = _workspace_fingerprint(args.cwd) + if workspace is None: + raise ValueError("complete worktree fingerprint unavailable") + if workspace["head"] != args.head: + raise ValueError("attestation HEAD mismatch") + if args.command == "inspect": + print(json.dumps({**vars(args), **workspace})) + return 0 + if workspace["fingerprint"] != args.confirm_fingerprint: + raise ValueError("explicit fingerprint confirmation does not match") + identity = subprocess.run(["gh", "api", "user", "--jq", ".login"], + capture_output=True, text=True, check=True).stdout.strip() + if not re.fullmatch(r"[A-Za-z0-9-]{1,100}", identity): + raise ValueError("operator identity unavailable") + now = datetime.now(timezone.utc).isoformat() + if args.command == "retry": + with closing(ledger.connect()) as connection, connection: + row = connection.execute( + "SELECT * FROM quota_history WHERE id=? AND repo_id=? AND executor=? AND retry_at IS NULL", + (args.record, args.repo, args.executor)).fetchone() + if row is None: + raise ValueError("quota record not found or retry already authorized") + updated = connection.execute( + "UPDATE quota_exhaustions SET expires_at=?, reason='retry_authorized' " + "WHERE repo_id=? AND executor=? AND model=? AND exhausted_at=?", + (now, args.repo, args.executor, row["model"], row["exhausted_at"])) + if updated.rowcount != 1: + raise ValueError("quota record was superseded") + connection.execute("UPDATE quota_history SET retry_at=? WHERE id=?", (now, args.record)) + record_id = args.record + else: + # Legacy attestation still requires historical quota evidence. It + # grants ownership only; it cannot assert provider availability. + with closing(ledger.connect()) as connection: + found = connection.execute( + "SELECT q.* FROM quota_exhaustions q WHERE repo_id=? AND executor=? AND model=? " + "AND NOT EXISTS (SELECT 1 FROM quota_history h WHERE h.repo_id=q.repo_id " + "AND h.executor=q.executor AND h.model=q.model AND h.exhausted_at=q.exhausted_at)", + (args.repo, args.original, args.model)).fetchone() + if not found: + raise ValueError("attestation requires a legacy quota record without automatic provenance history") + record_id = ledger.record( + repo=args.repo, task=args.task, stage=args.stage, pr=args.pr, + workspace=workspace, original=Target(args.original, "operator", args.model, "high"), + destination=Target(args.destination, "operator", args.model, "high"), + provenance="operator_attested", operator=identity) + record_event(store.path, repo_id=args.repo, task_id=args.task, + cycle_id=f"operator-{uuid.uuid4().hex}", kind="stage_result", + component="quota", status=args.command, detail=f"operator {identity}", + correlation={"record_id": record_id, "fingerprint": workspace["fingerprint"] if workspace else None}) + print(json.dumps({"record_id": record_id, "operator": identity, "action": args.command, "at": now})) + return 0 + except (ValueError, OSError, subprocess.SubprocessError, sqlite3.Error) as error: + parser.exit(2, f"quota recovery refused: {error}\n") + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/runtime.manifest b/scripts/runtime.manifest index d0867e2..e92ad7f 100644 --- a/scripts/runtime.manifest +++ b/scripts/runtime.manifest @@ -33,6 +33,7 @@ usage.py harness.py router.py telemetry.py +quota_state.py production_events.py cycle_status.py profile_config.py diff --git a/scripts/telemetry.py b/scripts/telemetry.py index 4cf3424..165b949 100644 --- a/scripts/telemetry.py +++ b/scripts/telemetry.py @@ -44,6 +44,11 @@ from datetime import datetime, timedelta, timezone from pathlib import Path +try: + from quota_state import QUOTA_SCHEMA, quota_reset +except ImportError: # pragma: no cover + from .quota_state import QUOTA_SCHEMA, quota_reset + try: from cycle_journal import JOURNAL_SCHEMA except ImportError: # pragma: no cover - imported as `scripts.telemetry` @@ -109,7 +114,9 @@ #: state could not be read. Both payload-only. #: 25: execution-plan revisions and check histories; verification_unverified #: is a distinct stop and cannot be cleared by generic CI success. -SCHEMA_VERSION = 25 +#: 26: append-only quota history and worktree provenance, independent of reset +#: eligibility. Provider prose remains exclusively in the redacted journal. +SCHEMA_VERSION = 26 APP_DIRNAME = "code-cycle-toolkit" DATABASE_NAME = "telemetry.sqlite" @@ -1311,6 +1318,7 @@ def __init__(self, path: Path | str | None = None, with closing(self._connect()) as connection: connection.executescript(SCHEMA) connection.executescript(JOURNAL_SCHEMA) + connection.executescript(QUOTA_SCHEMA) self._migrate_dispatch_attempts(connection) connection.commit() @@ -2099,38 +2107,28 @@ def attempt_updates(self, attempt_id: str) -> list[dict]: return result @staticmethod - def quota_expiry(detail: str | None, now: datetime | None = None) -> tuple[str, bool]: + def quota_expiry(detail: str | None, now: datetime | None = None) -> tuple[str | None, bool]: """Return a UTC expiry and whether the executor reported a reset time.""" - current = now or datetime.now(timezone.utc) - local_now = current.astimezone() - match = re.search(r"\b(\d{1,2}:\d{2}\s*[APap][Mm])\b", detail or "") - if match is not None and local_now.tzinfo is not None: - try: - reset_text = re.sub(r"(?i)(\d)(am|pm)$", r"\1 \2", match.group(1)) - reset = datetime.strptime(reset_text.upper(), "%I:%M %p") - candidate = local_now.replace( - hour=reset.hour, minute=reset.minute, second=0, microsecond=0, - ) - if candidate <= local_now: - candidate += timedelta(days=1) - return candidate.astimezone(timezone.utc).isoformat(), True - except (ValueError, OverflowError): - pass - return (current + timedelta(minutes=30)).isoformat(), False + expiry, reported, _ = quota_reset(detail, now) + return expiry, reported def record_quota_exhaustion(self, repo_id: str, executor: str, model: str, - detail: str | None = None) -> str: + detail: str | None = None) -> str | None: now = datetime.now(timezone.utc) - expires_at, reset_reported = self.quota_expiry(detail, now) + expires_at, reset_reported, reason = quota_reset(detail, now) with closing(self._connect()) as connection: + connection.execute( + "INSERT INTO quota_history (id,repo_id,executor,model,exhausted_at,expires_at,reason) " + "VALUES (?,?,?,?,?,?,?)", + (f"quota-{uuid.uuid4().hex}", repo_id, executor, model, now.isoformat(), expires_at, reason)) connection.execute( "INSERT INTO quota_exhaustions " "(repo_id, executor, model, exhausted_at, expires_at, reason, reset_reported) " - "VALUES (?, ?, ?, ?, ?, 'quota', ?) " + "VALUES (?, ?, ?, ?, ?, ?, ?) " "ON CONFLICT(repo_id, executor, model) DO UPDATE SET " "exhausted_at=excluded.exhausted_at, expires_at=excluded.expires_at, " "reason=excluded.reason, reset_reported=excluded.reset_reported", - (repo_id, executor, model, now.isoformat(), expires_at, int(reset_reported)), + (repo_id, executor, model, now.isoformat(), expires_at or "", reason, int(reset_reported)), ) connection.commit() return expires_at @@ -2138,14 +2136,9 @@ def record_quota_exhaustion(self, repo_id: str, executor: str, model: str, def active_quota_exhaustions(self, repo_id: str) -> list[dict]: now = datetime.now(timezone.utc).isoformat() with closing(self._connect()) as connection: - connection.execute( - "DELETE FROM quota_exhaustions WHERE repo_id=? AND expires_at<=?", - (repo_id, now), - ) - connection.commit() rows = connection.execute( "SELECT executor, model, exhausted_at, expires_at, reason, reset_reported " - "FROM quota_exhaustions WHERE repo_id=? AND expires_at>?", + "FROM quota_exhaustions WHERE repo_id=? AND (expires_at='' OR expires_at>?)", (repo_id, now), ).fetchall() return [dict(row) for row in rows] @@ -2153,8 +2146,8 @@ def active_quota_exhaustions(self, repo_id: str) -> list[dict]: def clear_quota_exhaustion(self, repo_id: str, executor: str, model: str) -> None: with closing(self._connect()) as connection: connection.execute( - "DELETE FROM quota_exhaustions WHERE repo_id=? AND executor=? AND model=?", - (repo_id, executor, model), + "UPDATE quota_exhaustions SET expires_at=? WHERE repo_id=? AND executor=? AND model=?", + (datetime.now(timezone.utc).isoformat(), repo_id, executor, model), ) connection.commit() diff --git a/scripts/windows-test-shards.json b/scripts/windows-test-shards.json index d26980b..a66d5b9 100644 --- a/scripts/windows-test-shards.json +++ b/scripts/windows-test-shards.json @@ -123,6 +123,7 @@ "test_manual_rereview", "test_native_prerequisites", "test_profile_config", + "test_quota_resume", "test_publication", "test_publication_audit", "test_review_contract", diff --git a/tests/test_cycle.py b/tests/test_cycle.py index 6be3d8c..6295754 100644 --- a/tests/test_cycle.py +++ b/tests/test_cycle.py @@ -861,7 +861,7 @@ def test_transient_failures_reroute_once_and_record_the_cause(self): self.assertEqual("not_started", row["payload"]["reroute_start_state"]) self.assertEqual(ex.Availability.READY, recorder.availability["codex"]) recorder.stage("implement", "later stage") - self.assertEqual(2, len(primary.dispatched)) + self.assertEqual(1 if code == "quota" else 2, len(primary.dispatched)) def test_dirty_quota_stop_is_reused_by_a_later_continue(self): clean = {"dirty": False, "workspace": {"head": "old"}, "remote": "old", @@ -907,6 +907,7 @@ def test_quota_stop_reports_recorded_reset_or_fallback_window(self): ) for detail, expected, guidance in cases: with self.subTest(detail=detail): + self.store.clear_quota_exhaustion("owner/repo", "codex", "gpt-6-luna") with patch("cycle.effect_snapshot", side_effect=[clean, dirty]): _, _, _, outcome = self.run_failure( code="quota", start="started", detail=detail, @@ -927,6 +928,9 @@ def test_success_clears_quota_only_in_production(self): self.assertTrue(calibration_outcome.succeeded, calibration_outcome.result.detail) self.assertEqual(1, len(self.store.active_quota_exhaustions("owner/repo"))) + # A production route now always honors active records, even without + # --continue. Explicit eligibility is needed before a successful retry. + self.store.clear_quota_exhaustion("owner/repo", "codex", "gpt-6-luna") production = self.recorder([ScriptedAdapter("codex")]) self.assertTrue(production.stage("implement", "production").succeeded) self.assertEqual([], self.store.active_quota_exhaustions("owner/repo")) diff --git a/tests/test_cycle_outcomes.py b/tests/test_cycle_outcomes.py index 8b25454..9f40fc6 100644 --- a/tests/test_cycle_outcomes.py +++ b/tests/test_cycle_outcomes.py @@ -185,17 +185,20 @@ class DispatchObservationTests(CycleTestCase): def test_fallbacks_and_contract_violations_are_counted(self) -> None: codex = ScriptedAdapter("codex", outcomes=[ (ex.DispatchOutcome.BLOCKED, "operating_quota"), + ]) + claude = ScriptedAdapter("claude", outcomes=[ + (ex.DispatchOutcome.SUCCEEDED, None), (ex.DispatchOutcome.CONTRACT_VIOLATION, None), ]) - recorder = self.recorder([codex, ScriptedAdapter("claude")]) + recorder = self.recorder([codex, claude]) recorder.stage("implement", "work") # quota, then the fallback recorder.availability["codex"] = ex.Availability.READY - recorder.stage("implement", "again") # the wrong model answered + recorder.stage("implement", "again") # recorded quota still routes to Claude recorder.close("HUMAN_INTERVENTION") outcome = self.store.cycle_outcome("owner/repo", recorder.cycle_id)["outcome"] - self.assertEqual(1, outcome["fallback_stages"]) + self.assertEqual(2, outcome["fallback_stages"]) self.assertEqual(1, outcome["contract_violations"]) diff --git a/tests/test_quota_resume.py b/tests/test_quota_resume.py new file mode 100644 index 0000000..f82f194 --- /dev/null +++ b/tests/test_quota_resume.py @@ -0,0 +1,436 @@ +"""Acceptance regressions for #389, including the real Git/CLI boundary.""" +from __future__ import annotations + +import sys +import unittest +import json +import os +import sqlite3 +import subprocess +import tempfile +from contextlib import closing +from datetime import datetime, timedelta, timezone +from pathlib import Path +from unittest.mock import patch +from zoneinfo import ZoneInfo, ZoneInfoNotFoundError + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) +from telemetry import Telemetry +from test_cycle import CycleTestCase, ScriptedAdapter +import cycle as cy +import executors as ex +import router +import quota_state as qs +from cycle_journal import Journal +from run_cycle import validate_start, check_change_request, CycleDriverError + + +class QuotaResumeTests(CycleTestCase): + def test_from_resume_skips_recorded_primary(self): + self.store.record_quota_exhaustion("owner/repo", "codex", "gpt-6-luna") + primary, fallback = ScriptedAdapter("codex"), ScriptedAdapter("claude") + recorder = self.recorder([primary, fallback], started_from="resolve", + change_request_id="384") + outcome = recorder.stage("resolve", "resume") + self.assertTrue(outcome.succeeded) + self.assertEqual([], primary.dispatched) + self.assertEqual(1, len(fallback.dispatched)) + self.assertIn("quota exhausted", outcome.decision.explain()) + + def test_resolve_escalation_cannot_restore_exhausted_target(self): + self.store.record_quota_exhaustion("owner/repo", "codex", "gpt-6-luna") + primary, fallback = ScriptedAdapter("codex"), ScriptedAdapter("claude") + recorder = self.recorder([primary, fallback], started_from="resolve", change_request_id="384") + outcome = recorder.stage("resolve", "resume", target_override=router.Target( + "codex", "openai", "gpt-6-luna", "high")) + self.assertTrue(outcome.succeeded) + self.assertFalse(primary.dispatched) + self.assertTrue(fallback.dispatched) + + +class ResetTests(unittest.TestCase): + def test_ordinal_date_and_timezone_are_not_dropped(self): + now = datetime(2026, 11, 2, 12, tzinfo=timezone.utc) + expiry, reported = Telemetry.quota_expiry( + "Usage limit; resets Nov 4th, 2026 3:20 PM UTC+01:00", now) + self.assertTrue(reported) + self.assertEqual("2026-11-04T14:20:00+00:00", expiry) + + def test_complete_dates_past_future_and_offsets(self): + now = datetime(2026, 11, 2, 12, tzinfo=timezone.utc) + for message, expected in [ + ("resets Nov 1st, 2026 3:20 PM UTC", "2026-11-01T15:20:00+00:00"), + ("resets November 4th 2026 15:20 UTC", "2026-11-04T15:20:00+00:00"), + ("resets 2026-11-04 15:20 -05:00", "2026-11-04T20:20:00+00:00"), + ("resets 2026-11-04T15:20:30Z", "2026-11-04T15:20:30+00:00"), + ("resets Nov 4th, 2026 3:20 PM", "2026-11-04T14:20:00+00:00"), + ]: + with self.subTest(message=message), patch("quota_state.local_zone", return_value=timezone(timedelta(hours=1))): + self.assertEqual((expected, True), Telemetry.quota_expiry(message, now)) + + def test_ambiguity_never_becomes_an_earlier_clock_only_reset(self): + for message in ("resets Nov 40th, 2026 3:20 PM", "resets Nov 4 3:20 PM", + "resets on a later date", "resets Nov 4th, 2026 3:20 PM CST", + "resets 2026-14-04 15:20 UTC", "resets Nov 4 2026 3:20 PM UTC+unknown", + "reset at 3:30pm (CST)", "reset at 3:30pm (America/Unknown)", + "reset at 3:30pm (UTC", "reset at 3:30pm UTC+01:00 extra", + "resets Tuesday 3:20 PM", "resets Tue 3:20 PM", + "resets tomorrow 3:20 PM", "Try again in 4 hours 23 minutes", + "Try again in 2 days", "Try again after 45 minutes", + "Try again in 4h23m", "Try again in four hours"): + with self.subTest(message=message): + self.assertEqual((None, True), Telemetry.quota_expiry(message)) + + def test_dst_repeated_hour_chooses_latest_interpretation(self): + try: + ZoneInfo("Europe/Madrid") + except ZoneInfoNotFoundError: + self.skipTest("IANA timezone database is unavailable") + with patch.dict(os.environ, {"TZ": "Europe/Madrid"}): + expiry, _ = Telemetry.quota_expiry("resets Oct 25th, 2026 2:30 AM") + self.assertEqual("2026-10-25T01:30:00+00:00", expiry) + + def test_missing_reset_uses_only_minimum_window(self): + now = datetime(2026, 11, 2, 12, tzinfo=timezone.utc) + self.assertEqual(("2026-11-02T12:30:00+00:00", False), + Telemetry.quota_expiry("quota exceeded", now)) + + def test_platform_local_clock_without_iana_zone(self): + now = datetime(2026, 11, 2, 12, tzinfo=timezone.utc) + for message, hour, minute in [("quota exhausted; reset at 9:46 PM", 21, 46), + ("resets 2:50pm", 14, 50)]: + wall = now.astimezone().replace(tzinfo=None, hour=hour, minute=minute) + expected = max(wall.replace(fold=f).astimezone(timezone.utc) for f in (0, 1)) + if expected <= now: + wall += timedelta(days=1) + expected = max(wall.replace(fold=f).astimezone(timezone.utc) for f in (0, 1)) + with self.subTest(message=message), patch.dict(os.environ, {"TZ": ""}), patch( + "quota_state.Path.open", side_effect=FileNotFoundError): + self.assertEqual((expected.isoformat(), True), Telemetry.quota_expiry(message, now)) + + def test_platform_local_dated_reset_uses_rules_for_that_date(self): + wall = datetime(2026, 11, 4, 15, 20) + expected = max(wall.replace(fold=f).astimezone(timezone.utc) for f in (0, 1)) + with patch("quota_state.local_zone", return_value=None): + self.assertEqual((expected.isoformat(), True), Telemetry.quota_expiry( + "resets Nov 4th, 2026 3:20 PM", datetime(2026, 7, 2, tzinfo=timezone.utc))) + + def test_parenthesized_explicit_zones(self): + now = datetime(2026, 11, 2, 12, tzinfo=timezone.utc) + for suffix, expected in [("(UTC)", "2026-11-02T15:30:00+00:00"), + ("(UTC-08:00)", "2026-11-02T23:30:00+00:00")]: + with self.subTest(suffix=suffix), patch("quota_state.local_zone", return_value=timezone(timedelta(hours=1))): + self.assertEqual((expected, True), Telemetry.quota_expiry("reset at 3:30pm " + suffix, now)) + try: + ZoneInfo("America/Los_Angeles") + except ZoneInfoNotFoundError: + self.skipTest("IANA timezone database is unavailable") + with patch("quota_state.local_zone", return_value=timezone(timedelta(hours=1))): + self.assertEqual(("2026-11-02T23:30:00+00:00", True), Telemetry.quota_expiry( + "reset at 3:30pm (America/Los_Angeles)", now)) + + +class GitHandoffTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + # Cleanups run last-in first-out, so this check precedes the removal. + self.addCleanup(self.assert_no_open_handles) + self.root = Path(self.temp.name) + self.cwd = self.root / "checkout" + self.cwd.mkdir() + self.git("init", "-b", "task-330") + self.git("config", "user.email", "fixture@example.invalid") + self.git("config", "user.name", "Fixture") + for name in ("one.py", "two.py"): + (self.cwd / name).write_bytes(b"initial\n") + self.git("add", ".") + self.git("commit", "-m", "fixture") + self.head = self.git("rev-parse", "HEAD") + self.store = Telemetry(self.root / "telemetry.sqlite") + self.snapshots = patch("cycle.effect_snapshot", side_effect=self.snapshot) + self.snapshots.start() + self.addCleanup(self.snapshots.stop) + + def assert_no_open_handles(self): + """Windows cannot delete a file that still has an open handle, so a + leaked SQLite connection fails the temporary-directory cleanup there. + Where the process exposes its descriptors, fail a handle that is still + open here too. Since Python 3.11 an unclosed connection can outlive its + last name until the cyclic collector runs, so close every connection.""" + descriptors = Path("/proc/self/fd") + if not descriptors.is_dir(): + return + root = str(self.root.resolve()) + os.sep + held = [] + for descriptor in descriptors.iterdir(): + try: + target = os.readlink(descriptor) + except OSError: + continue + if target.startswith(root): + held.append(target) + self.assertEqual([], held, "handles left open in the fixture directory") + + def database(self): + """Open the fixture database; the caller's ``with`` closes it.""" + return closing(sqlite3.connect(self.store.path)) + + def execute(self, statement): + with self.database() as connection, connection: + connection.execute(statement) + + def git(self, *args): + return subprocess.run(["git", *args], cwd=self.cwd, capture_output=True, + text=True, check=True).stdout.strip() + + def snapshot(self, *args, **kwargs): + return {"dirty": bool(self.git("status", "--porcelain", "--untracked-files=all")), + "workspace": ex._workspace_fingerprint(str(self.cwd)), "remote": "same", + "pulls": [], "observed_pulls": {}, "comments": []} + + def recorder(self, adapters, **kw): + recorder = cy.CycleRecorder( + self.store, "owner/repo", "330", router.TaskSignals(difficulty=3), + availability={a.name: ex.Availability.READY for a in adapters}, + registry=ex.Registry(adapters), started_from="resolve", change_request_id="384", **kw) + recorder.journal = Journal(self.store.path, "owner/repo", "330", recorder.cycle_id, + change_request_id="384") + return recorder + + def edit_three_files(self): + (self.cwd / "one.py").write_bytes(b"staged\x00bytes\n") + self.git("add", "one.py") + (self.cwd / "one.py").write_bytes(b"unstaged after staged\xff\n") + (self.cwd / "two.py").write_bytes(b"second change\n") + (self.cwd / "three.py").write_bytes(b"untracked\x00\xff\n") + + def exhaust(self, message="quota exceeded"): + fixture = self + class Exhausted(ScriptedAdapter): + def dispatch(self, target, task, **kw): + fixture.edit_three_files() + return ex.DispatchResult(ex.DispatchOutcome.FAILED, self.name, target, + error_code="quota", start_state="started", + process_exited=True, detail=message) + primary, fallback = Exhausted("codex"), ScriptedAdapter("claude") + outcome = self.recorder([primary, fallback]).stage("resolve", "work", cwd=str(self.cwd)) + self.assertFalse(outcome.succeeded) + self.assertEqual([], fallback.dispatched) + assert outcome.result is not None + self.assertIn("dirty_workspace", outcome.result.detail) + return outcome + + def rows(self, table): + with self.database() as connection: + connection.row_factory = sqlite3.Row + return [dict(row) for row in connection.execute(f"SELECT * FROM {table}")] + + def resume(self, **kw): + primary, fallback = ScriptedAdapter("codex"), ScriptedAdapter("claude") + outcome = self.recorder([primary, fallback]).stage("resolve", "resume", cwd=str(self.cwd), **kw) + return outcome, primary, fallback + + def test_dirty_fallback_receives_exact_bytes_and_head_with_audit(self): + stopped = self.exhaust() + before = ex._workspace_fingerprint(str(self.cwd)) + assert before is not None + outcome, primary, fallback = self.resume() + self.assertTrue(outcome.succeeded) + self.assertFalse(primary.dispatched) + self.assertEqual(1, len(fallback.dispatched)) + self.assertIn("inspect the existing", fallback.dispatched[0]) + self.assertEqual(before, ex._workspace_fingerprint(str(self.cwd))) + self.assertEqual(self.head, self.git("rev-parse", "HEAD")) + row = self.rows("quota_handoffs")[0] + self.assertEqual(("resolve", "384", "codex", "claude", before["fingerprint"]), + (row["stage"], row["pr"], row["original_executor"], + row["destination_executor"], row["fingerprint"])) + self.assertIsNotNone(row["used_at"]) + # Stop advice is accepted by the real CLI parser and start validator. + assert stopped.result is not None + self.assertIn("--from resolve --pr 384", stopped.result.detail) + validate_start("resolve", "384") + + def test_altered_staged_unstaged_or_untracked_diff_is_refused(self): + for kind in ("staged", "unstaged", "untracked"): + with self.subTest(kind=kind): + # Each mutation is reset only inside this disposable fixture. + if self.git("status", "--porcelain"): + self.git("reset", "--hard", self.head) + self.git("clean", "-fd") + self.execute("DELETE FROM quota_exhaustions") + self.exhaust() + path = self.cwd / ("three.py" if kind == "untracked" else "two.py") + path.write_bytes(b"operator changed it\n") + if kind == "staged": + self.git("add", "two.py") + before = ex._workspace_fingerprint(str(self.cwd)) + outcome, primary, fallback = self.resume() + self.assertTrue(outcome.decision.blocked) + self.assertIn("fingerprint mismatch", outcome.decision.explain()) + self.assertFalse(primary.dispatched or fallback.dispatched) + self.assertEqual(before, ex._workspace_fingerprint(str(self.cwd))) + + def test_unrelated_dirty_resume_has_no_ownership(self): + self.edit_three_files() + self.store.record_quota_exhaustion("owner/repo", "codex", "gpt-6-luna") + outcome, primary, fallback = self.resume() + self.assertIn("no quota worktree provenance", outcome.decision.explain()) + self.assertFalse(primary.dispatched or fallback.dispatched) + + def test_expired_window_returns_to_original_executor_keeps_provenance(self): + self.exhaust() + self.execute("UPDATE quota_exhaustions SET expires_at='2000-01-01T00:00:00+00:00'") + self.assertEqual([], self.store.active_quota_exhaustions("owner/repo")) + self.assertEqual(1, len(self.rows("quota_handoffs"))) + before = ex._workspace_fingerprint(str(self.cwd)) + outcome, primary, fallback = self.resume() + self.assertTrue(outcome.succeeded) + self.assertEqual(1, len(primary.dispatched)) + self.assertFalse(fallback.dispatched) + self.assertEqual(before, ex._workspace_fingerprint(str(self.cwd))) + self.assertEqual(1, len(self.rows("quota_history"))) + + def test_repeated_exhaustion_after_expiry_records_new_checkpoint(self): + self.exhaust() + self.execute("UPDATE quota_exhaustions SET expires_at='2000-01-01T00:00:00+00:00'") + self.exhaust() + self.assertEqual(2, len(self.rows("quota_history"))) + self.assertEqual(2, len(self.rows("quota_handoffs"))) + outcome, primary, fallback = self.resume() + self.assertTrue(outcome.succeeded) + self.assertFalse(primary.dispatched) + self.assertTrue(fallback.dispatched) + self.assertTrue(all(row["used_at"] for row in self.rows("quota_handoffs"))) + + def test_wrong_stage_pr_and_original_model_cannot_match(self): + self.exhaust() + ledger = qs.Handoffs(self.store.path) + values = dict(repo="owner/repo", task="330", stage="resolve", pr="384", + workspace=ex._workspace_fingerprint(str(self.cwd)), + destination=router.Target("codex", "openai", "gpt-6-luna", "high")) + for override in ({"stage": "rereview"}, {"pr": "999"}, {"task": "999"}, + {"destination": router.Target("codex", "openai", "gpt-6.1-sol", "high")}): + with self.subTest(override=override), self.assertRaises(ValueError): + ledger.match(**{**values, **override}) + + def test_ordinary_dirty_executor_change_is_refused(self): + self.edit_three_files() + outcome, primary, fallback = self.resume() + self.assertTrue(outcome.decision.blocked) + self.assertFalse(primary.dispatched or fallback.dispatched) + + def test_head_mismatch_is_still_refused(self): + self.exhaust() + self.git("add", ".") + self.git("commit", "-m", "forbidden WIP fixture") + (self.cwd / "two.py").write_bytes(b"dirty again\n") + outcome, primary, fallback = self.resume() + self.assertIn("HEAD mismatch", outcome.decision.explain()) + self.assertFalse(primary.dispatched or fallback.dispatched) + + def test_ambiguous_reset_keeps_redacted_message_out_of_telemetry(self): + stopped = self.exhaust("quota exceeded; resets Nov 40th, 2026 3:20 PM; token=ghp_123456789012345678901234567890123456") + assert stopped.result is not None + self.assertNotIn("ghp_123456789012345678901234567890123456", stopped.result.detail) + self.assertIsNone(self.rows("quota_history")[0]["expires_at"]) + self.assertEqual("reset_time_ambiguous", self.rows("quota_history")[0]["reason"]) + journal = json.dumps(self.rows("journal_events")) + self.assertIn("Nov 40th", journal) + self.assertNotIn("ghp_123456789012345678901234567890123456", journal) + self.assertNotIn("Nov 40th", json.dumps(self.rows("quota_exhaustions"))) + outcome, primary, fallback = self.resume() + self.assertTrue(outcome.succeeded) + self.assertFalse(primary.dispatched) + + def test_operator_retry_names_record_and_preserves_history(self): + self.store.record_quota_exhaustion("owner/repo", "codex", "gpt-6-luna", "reset later") + record_id = self.rows("quota_history")[0]["id"] + args = ["--database", str(self.store.path), "--repo", "owner/repo", "--task", "330", + "retry", "--record", record_id, "--executor", "codex"] + with patch("quota_state.subprocess.run", return_value=subprocess.CompletedProcess([], 0, "operator-login\n")): + self.assertEqual(0, qs.main(args)) + self.assertEqual([], self.store.active_quota_exhaustions("owner/repo")) + self.assertIsNotNone(self.rows("quota_history")[0]["retry_at"]) + self.assertEqual(1, len(self.rows("quota_exhaustions"))) + with patch("quota_state.subprocess.run", return_value=subprocess.CompletedProcess([], 0, "operator-login\n")): + with self.assertRaises(SystemExit) as error: + qs.main(args) + self.assertEqual(2, error.exception.code) + + def test_legacy_attestation_requires_exact_confirmation_and_is_single_use(self): + self.edit_three_files() + self.store.record_quota_exhaustion("owner/repo", "codex", "gpt-6-luna") + # Model a pre-#389 record, which has no new exhaustion history. + self.execute("DELETE FROM quota_history") + workspace = ex._workspace_fingerprint(str(self.cwd)) + assert workspace is not None + fingerprint = workspace["fingerprint"] + args = ["--database", str(self.store.path), "--repo", "owner/repo", "--task", "330", + "attest", "--cwd", str(self.cwd), "--pr", "384", "--stage", "resolve", + "--original", "codex", "--model", "gpt-6-luna", "--destination", "claude", + "--head", self.head, "--confirm-fingerprint"] + with self.assertRaises(SystemExit): + qs.main(args + ["0" * 64]) + self.assertEqual([], self.rows("quota_handoffs")) + run = subprocess.run + def operator_run(command, **kwargs): + if command[:3] == ["gh", "api", "user"]: + return subprocess.CompletedProcess(command, 0, "operator-login\n") + return run(command, **kwargs) + with patch("quota_state.subprocess.run", side_effect=operator_run): + self.assertEqual(0, qs.main(args + [fingerprint])) + row = self.rows("quota_handoffs")[0] + self.assertEqual(("operator_attested", "operator-login"), (row["provenance"], row["operator"])) + outcome, primary, fallback = self.resume() + self.assertTrue(outcome.succeeded) + self.assertFalse(primary.dispatched) + self.assertIn("operator_attested", fallback.dispatched[0]) + outcome, primary, fallback = self.resume() + self.assertTrue(outcome.decision.blocked) + + def test_cli_help_accepts_operator_commands(self): + for command in ("retry", "inspect", "attest"): + done = subprocess.run([sys.executable, str(Path(qs.__file__)), + "--database", str(self.store.path), "--repo", "owner/repo", + "--task", "330", command, "--help"], capture_output=True, text=True) + self.assertEqual(0, done.returncode, done.stderr) + + def test_new_unattested_work_cannot_use_legacy_attestation(self): + self.edit_three_files() + self.store.record_quota_exhaustion("owner/repo", "codex", "gpt-6-luna") + workspace = ex._workspace_fingerprint(str(self.cwd)) + assert workspace is not None + fingerprint = workspace["fingerprint"] + args = ["--database", str(self.store.path), "--repo", "owner/repo", "--task", "330", + "attest", "--cwd", str(self.cwd), "--pr", "384", "--stage", "resolve", + "--original", "codex", "--model", "gpt-6-luna", "--destination", "claude", + "--head", self.head, "--confirm-fingerprint", fingerprint] + run = subprocess.run + def operator_run(command, **kwargs): + if command[:3] == ["gh", "api", "user"]: + return subprocess.CompletedProcess(command, 0, "operator-login\n") + return run(command, **kwargs) + with patch("quota_state.subprocess.run", side_effect=operator_run), self.assertRaises(SystemExit) as error: + qs.main(args) + self.assertEqual(2, error.exception.code) + self.assertEqual([], self.rows("quota_handoffs")) + + def test_stop_advice_is_accepted_at_the_real_cli_boundary(self): + self.exhaust() + script = Path(__file__).resolve().parents[1] / "scripts" / "run_cycle.py" + done = subprocess.run([sys.executable, str(script), "--repo", "owner/repo", + "--task", "330", "--from", "resolve", "--pr", "384", + "--cwd", str(self.cwd), "--dry-run"], + cwd=script.parents[1], capture_output=True, text=True) + self.assertEqual(0, done.returncode, done.stderr) + + def test_existing_pr_head_guard_remains_effective(self): + def run(command, **kw): + if command[0] == "gh": + body = {"state": "OPEN", "headRefName": "task-330", "headRefOid": "f" * 40} + return subprocess.CompletedProcess(command, 0, json.dumps(body)) + return subprocess.run(command, **kw) + with self.assertRaisesRegex(CycleDriverError, "not at"): + check_change_request("owner/repo", "384", str(self.cwd), run=run) diff --git a/tests/test_work_item_outcomes.py b/tests/test_work_item_outcomes.py index 38130e8..5ebd625 100644 --- a/tests/test_work_item_outcomes.py +++ b/tests/test_work_item_outcomes.py @@ -70,7 +70,7 @@ def test_schema_fourteen_records_transitions_and_deduplicates_snapshots(self) -> self.assertEqual("unchanged", repeated["reason"]) self.assertFalse(stale["created"]) self.assertEqual("stale", stale["reason"]) - self.assertEqual(25, tm.SCHEMA_VERSION) + self.assertEqual(26, tm.SCHEMA_VERSION) self.assertEqual(1, len(self.store.work_item_disposition_events("owner/repo"))) event = self.store.latest_work_item_disposition("owner/repo", "github", "owner/repo", "120") self.assertEqual(["cycle-1"], event["cycle_ids"])