diff --git a/activity-definitions/repo-manager-bounded-freshness.md b/activity-definitions/repo-manager-bounded-freshness.md new file mode 100644 index 0000000..277b3a9 --- /dev/null +++ b/activity-definitions/repo-manager-bounded-freshness.md @@ -0,0 +1,63 @@ +--- +id: repo-manager-bounded-freshness +name: Repo Manager Bounded Freshness +type: activity-definition +version: "0.1" +enabled: false +owner: repo-manager +governance: custodian +status: proposed +trigger: + type: cron + cron_expression: "40 */2 * * *" + timezone: UTC + misfire_policy: skip +context_sources: + - type: repo-manager + query: freshness_run + operation: repo_manager_freshness + required: true + params: + limit: 6 + max_seconds: 240 + apply: true + bind_to: context.freshness +--- + +# Repo Manager Bounded Freshness + +Draft from RMGR-WP-0020-T03, tracked in ACTIVITY-WP-0044. **Not activated.** +`enabled: false` prevents any Temporal schedule; the founder flips it only after +the RMGR-WP-0020-T04 activation steps and T05 proof. + +Each fire runs one bounded operation `repo_manager_freshness` +(ACT-ADR-007 registry), which invokes the Repo Manager CLI with a code-fixed argv: + +```text +rmgr freshness run --limit 6 --max-seconds 240 +``` + +(`--dry-run` is appended when `apply: false`.) Repo Manager owns repository +selection, Forge HEAD resolution, reconciliation from immutable Forge commits +and the durable `repo-manager.freshness-receipt.v1` receipts. Activity Core +persists only a counted projection (receipt count per outcome, exit code). + +- Every 2 h at :40 UTC; 141 repositories / 6 per fire x 12 fires/day is about a + two-day rotation. Target SLA once enabled: every admissible repository has a + receipt younger than 48 h. +- Single Temporal attempt, timeout 300 s (240 s CLI budget + grace); misfires + are skipped, never replayed. +- No workstation checkout reads (ADR-002), no legacy `--all --remote` sweep, no + `for_each` task spawning; the rule below never emits tasks. + +Rollback: set `enabled: false` (or `status: paused`). + +```rule +id: repo-manager-bounded-freshness-no-tasks +condition: "False" +action: + task_template: Repo Manager freshness (operation-only; never rule-emitted) + description: The bounded operation does the work; this rule never emits tasks. + priority: low + labels: ["automated", "freshness"] +``` diff --git a/docs/owner-return-pickup-activation-plan.md b/docs/owner-return-pickup-activation-plan.md index e4c9c07..0cea396 100644 --- a/docs/owner-return-pickup-activation-plan.md +++ b/docs/owner-return-pickup-activation-plan.md @@ -21,6 +21,8 @@ STATE-WP-0094-T03. Nothing below has been applied. 2. A caller that builds `current_tasks` from fresh authoritative source bound to the scanned repository revision, resolving every predecessor and decision (contract: "Current eligibility guard"). No manufactured empty predecessor lists. + Source builder: `activity_core.owner_return_eligibility.build_current_tasks` + (ACTIVITY-WP-0043-T05); the C-25 caller must pass the scanned commit SHA. 3. The caller runs inside the existing supported runner (Custodian dependency monitor / established Activity Core worker); no new schedule is introduced. 4. Founder review of this plan and of the definition/grant above. diff --git a/docs/platform-recovery-scheduling-plan.md b/docs/platform-recovery-scheduling-plan.md index cbabb26..a5ce0a3 100644 --- a/docs/platform-recovery-scheduling-plan.md +++ b/docs/platform-recovery-scheduling-plan.md @@ -25,7 +25,8 @@ schedule, retention action, or production rollout follows from role acceptance. Scaleway physical recovery or Forgejo archive recovery evidence. Its old host kubeconfig names and RPO wording are not proof of current cluster routing or an accepted guarantee; review those before reusing the wiring. -- `src/activity_core/bounded_operations.py` permits exactly three operations. +- `src/activity_core/bounded_operations.py` permits exactly four operations + (the fourth, `repo_manager_freshness`, backs a disabled draft; ACTIVITY-WP-0044). CNPG backup has one Temporal attempt because no durable remote idempotency receipt is available. New restore dispatch must not inherit retry defaults. - The stale-token blocker was resolved under RPF-WP-0045. Read-only inspection diff --git a/src/activity_core/activities.py b/src/activity_core/activities.py index 0db2127..536ce43 100644 --- a/src/activity_core/activities.py +++ b/src/activity_core/activities.py @@ -254,6 +254,10 @@ async def execute_bounded_operation(payload: dict[str, Any]) -> dict[str, Any]: from activity_core.context_resolvers.cnpg_backup import cnpg_option_a_backup result = cnpg_option_a_backup(params) + elif spec.operation_id == "repo_manager_freshness": + from activity_core.context_resolvers.repo_manager import repo_manager_freshness + + result = repo_manager_freshness(params) else: raise ApplicationError( f"bounded operation {spec.operation_id!r} uses a dedicated activity", diff --git a/src/activity_core/bounded_operations.py b/src/activity_core/bounded_operations.py index 32a97b2..63d5a8b 100644 --- a/src/activity_core/bounded_operations.py +++ b/src/activity_core/bounded_operations.py @@ -60,6 +60,19 @@ _SPECS = ( credential_route="railiance-cnpg-option-a-backup", evidence_mode="context-evidence-sink", ), + # RMGR-WP-0020-T03 / ACTIVITY-WP-0044: Repo Manager owns the freshness + # engine; Activity Core only invokes its fixed CLI with bounded arguments. + BoundedOperationSpec( + operation_id="repo_manager_freshness", + source_type="repo-manager", + query="freshness_run", + resolve_before_execute=False, + max_timeout_seconds=300, + temporal_max_attempts=1, + idempotency="single Temporal attempt; per-commit projections are idempotent (RMGR-WP-0017)", + credential_route="repo-manager-freshness-runner", + evidence_mode="repo-manager-freshness-receipt", + ), ) BOUNDED_OPERATION_REGISTRY = {spec.operation_id: spec for spec in _SPECS} @@ -80,6 +93,12 @@ _CANONICAL_SCRIPT_PATHS = { "forgejo_package_prune": "/opt/railiance-platform/tools/cmd/forgejo-package-prune", "cnpg_option_a_backup": "/opt/railiance-platform/tools/cmd/cnpg-option-a-backup", } +# repo-manager.freshness-receipt.v1 outcomes (RMGR-WP-0020-T01). +REPO_MANAGER_FRESHNESS_OUTCOMES = ( + "applied", "noop", "queued", "failed", "refused", "skipped", +) +REPO_MANAGER_FRESHNESS_MAX_LIMIT = 12 +REPO_MANAGER_FRESHNESS_MAX_SECONDS = 270 _TARGET_RE = re.compile(r"^[A-Za-z0-9_.-]+$") @@ -136,6 +155,26 @@ def normalize_operation_result(operation_id: str, raw: Any) -> dict[str, Any]: "failed": _safe_count(result.get("failed")), "script_exit_code": exit_code, } + if operation_id == "repo_manager_freshness": + receipts = result.get("receipts") + outcomes = {name: 0 for name in REPO_MANAGER_FRESHNESS_OUTCOMES} + if isinstance(receipts, list): + for receipt in receipts: + outcome = receipt.get("outcome") if isinstance(receipt, dict) else None + if outcome in outcomes: + outcomes[outcome] += 1 + exit_code = result.get("exit_code", 0) + if isinstance(exit_code, bool) or not isinstance(exit_code, int): + exit_code = 1 + return { + "kind": operation_id, + "apply": bool(result.get("apply")), + "limit": _safe_count(result.get("limit")), + "max_seconds": _safe_count(result.get("max_seconds")), + "receipt_count": len(receipts) if isinstance(receipts, list) else 0, + "outcomes": outcomes, + "exit_code": exit_code, + } raise ValueError(f"no result projection for bounded operation {operation_id!r}") @@ -192,6 +231,8 @@ def validate_bounded_operations( _validate_prune(params) elif spec.operation_id == "cnpg_option_a_backup": _validate_backup(params) + elif spec.operation_id == "repo_manager_freshness": + _validate_repo_manager_freshness(params) def _validate_timeout(spec: BoundedOperationSpec, params: dict[str, Any]) -> None: @@ -259,6 +300,36 @@ def _validate_backup(params: dict[str, Any]) -> None: _require_context_evidence(params, "cnpg_option_a_backup") +def _validate_repo_manager_freshness(params: dict[str, Any]) -> None: + _explicit_bool(params, "apply", "repo_manager_freshness") + limit = params.get("limit") + if ( + isinstance(limit, bool) + or not isinstance(limit, int) + or not 1 <= limit <= REPO_MANAGER_FRESHNESS_MAX_LIMIT + ): + raise ValueError( + "bounded operation repo_manager_freshness limit must be in " + f"1..{REPO_MANAGER_FRESHNESS_MAX_LIMIT}" + ) + max_seconds = params.get("max_seconds") + if ( + isinstance(max_seconds, bool) + or not isinstance(max_seconds, int) + or not 1 <= max_seconds <= REPO_MANAGER_FRESHNESS_MAX_SECONDS + ): + raise ValueError( + "bounded operation repo_manager_freshness max_seconds must be in " + f"1..{REPO_MANAGER_FRESHNESS_MAX_SECONDS}" + ) + for forbidden in ("command", "argv", "rmgr_bin", "script"): + if forbidden in params: + raise ValueError( + f"bounded operation repo_manager_freshness must not declare {forbidden}; " + "the CLI invocation is code-fixed" + ) + + def _validate_canonical_script( params: dict[str, Any], field: str, operation_id: str ) -> None: diff --git a/src/activity_core/context_resolvers/repo_manager.py b/src/activity_core/context_resolvers/repo_manager.py new file mode 100644 index 0000000..4c5620a --- /dev/null +++ b/src/activity_core/context_resolvers/repo_manager.py @@ -0,0 +1,63 @@ +"""Repo Manager bounded freshness invocation (RMGR-WP-0020-T03, ACTIVITY-WP-0044). + +Invokes the Repo Manager CLI ``rmgr freshness run --limit N --max-seconds S +[--dry-run]`` (RMGR-WP-0020-T02 interface). The argv is code-fixed: a +definition supplies only the validated ``limit``, ``max_seconds`` and ``apply`` +values. The executable is ``rmgr`` on PATH unless the worker sets +``ACTIVITY_CORE_RMGR_BIN``; definitions cannot choose it. Stdout is expected to +be a JSON object carrying ``receipts`` (``repo-manager.freshness-receipt.v1``); +only the counted projection in ``bounded_operations`` is persisted. +""" + +from __future__ import annotations + +import json +import os +import subprocess +from typing import Any + +# Grace on top of the CLI's own --max-seconds budget for start-up/teardown. +_TIMEOUT_GRACE_SECONDS = 30 + + +def build_freshness_argv(params: dict[str, Any]) -> list[str]: + limit = int(params["limit"]) + max_seconds = int(params["max_seconds"]) + argv = [ + os.environ.get("ACTIVITY_CORE_RMGR_BIN", "rmgr"), + "freshness", + "run", + "--limit", + str(limit), + "--max-seconds", + str(max_seconds), + ] + if params.get("apply") is not True: + argv.append("--dry-run") + return argv + + +def repo_manager_freshness(params: dict[str, Any]) -> dict[str, Any]: + """Run one bounded freshness pass and return its JSON summary.""" + argv = build_freshness_argv(params) + timeout = int(params["max_seconds"]) + _TIMEOUT_GRACE_SECONDS + completed = subprocess.run( + argv, capture_output=True, text=True, timeout=timeout, check=False + ) + stdout = (completed.stdout or "").strip() + if completed.returncode != 0 and not stdout: + raise RuntimeError( + f"repo_manager_freshness failed (exit {completed.returncode}): " + f"{(completed.stderr or '').strip()[:300]}" + ) + try: + summary = json.loads(stdout) + except json.JSONDecodeError as exc: + raise RuntimeError(f"repo_manager_freshness returned invalid JSON: {exc}") from exc + if not isinstance(summary, dict): + raise RuntimeError("repo_manager_freshness output must be a JSON object") + summary.setdefault("apply", params.get("apply") is True) + summary.setdefault("limit", int(params["limit"])) + summary.setdefault("max_seconds", int(params["max_seconds"])) + summary["exit_code"] = completed.returncode + return summary diff --git a/src/activity_core/ops_console.py b/src/activity_core/ops_console.py index 7a3418d..f42a08f 100644 --- a/src/activity_core/ops_console.py +++ b/src/activity_core/ops_console.py @@ -40,6 +40,7 @@ _AUDIT_BUFFER: deque[dict[str, Any]] = deque(maxlen=100) SIDE_EFFECT_MARKERS = ( "forgejo_package_prune", "cnpg_option_a_backup", + "repo_manager_freshness", "apply: true", '"apply": true', "'apply': true", diff --git a/src/activity_core/owner_return_eligibility.py b/src/activity_core/owner_return_eligibility.py new file mode 100644 index 0000000..704299e --- /dev/null +++ b/src/activity_core/owner_return_eligibility.py @@ -0,0 +1,474 @@ +"""Eligibility input builder for owner-return pickup (ACTIVITY-WP-0043-T05). + +Builds the ``current_tasks`` mapping consumed by State Hub's offline planner +(``state-hub/scripts/owner_return_handoff.py::plan_owner_returns``) from fresh +authoritative source, per the contract in ``state-hub/docs/owner-return-handoff.md`` +("Current eligibility guard"). + +Source binding +-------------- +The caller supplies ``scan_revision``: the full 40-hex commit SHA that the C-25 +consistency scan read. Every workplan file is read from the Forge at exactly +that commit (``/contents`` and ``/raw`` with ``ref=``), i.e. immutable +commit content -- never a workstation checkout (ADR-002) and never a branch +name that could move between reads. Before reading, the builder verifies that +the Forge knows the commit (``/git/commits/`` echoes the same SHA) and, by +default, that it is still the default-branch HEAD (``require_current_head``): +a scan of an older revision is stale and refuses rather than plan from it. + +Resolution rules (fail closed) +------------------------------ +* Missing or malformed authoritative fields (task ``status``, workplan + ``status``, non-boolean ``needs_human``, bad ``wait_until``, malformed + ``depends_on``) raise :class:`EligibilityError` -- no partial mapping. +* Duplicate record identities at the revision raise (no winner is chosen). +* Predecessors: task-block ``depends_on`` plus the owning workplan's + frontmatter dependencies (a workplan-level gate applies to all its tasks). + Targets are resolved from the same commit; cross-repository targets only + through an explicit ``external_status`` resolver. A target that cannot be + resolved is recorded as ``None`` (unknown) -- never dropped, never replaced + by an empty list. +* Decisions are hub-native records: ``decision_status`` comes from + ``decision_status`` (e.g. State Hub ``GET /decisions/{id}``); lookup failure + yields ``None`` (unknown), which the planner stops on. + +Nothing here writes, schedules or dispatches. The HTTP client is injected; the +caller owns any read credential. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from datetime import date +from typing import Any, Callable, Mapping, Sequence +from urllib.parse import quote + +import httpx +import yaml + +SCHEMA = "activity-core.owner-return-eligibility.v1" +DEFAULT_FORGE_API = "https://forgejo.coulomb.social/api/v1" +DEFAULT_FORGE_OWNER = "coulomb" +WORKPLAN_DIRS = ("workplans", "workplans/archived") + +_SHA = re.compile(r"^[0-9a-f]{40}$") +_REPO = re.compile(r"^[a-z0-9][a-z0-9._-]{0,127}$") +_TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL) +_SHORT_TASK_RE = re.compile(r"^T\d{2,3}$", re.IGNORECASE) +_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$") +_EMPTY_FLAVOR = {"", "~", "null", "none", "nil"} + +StatusResolver = Callable[[str], "str | None"] + + +class EligibilityError(ValueError): + """Authoritative eligibility could not be established; refuse planning.""" + + +# --------------------------------------------------------------------------- # +# Forge reads pinned to one commit +# --------------------------------------------------------------------------- # + + +class ForgeCommitReader: + """Read-only Forgejo API access pinned to a single immutable commit.""" + + def __init__( + self, + client: httpx.Client, + repo: str, + *, + owner: str = DEFAULT_FORGE_OWNER, + api_base: str = DEFAULT_FORGE_API, + ) -> None: + if not _REPO.fullmatch(repo): + raise EligibilityError("invalid repository identity") + self._client = client + self._repo_url = f"{api_base.rstrip('/')}/repos/{quote(owner)}/{quote(repo)}" + + def _get(self, path: str, **params: str) -> httpx.Response: + try: + response = self._client.get(f"{self._repo_url}{path}", params=params or None) + except httpx.HTTPError as exc: + raise EligibilityError(f"forge read failed: {type(exc).__name__}") from exc + return response + + def _json(self, path: str, **params: str) -> Any: + response = self._get(path, **params) + if response.status_code != 200: + raise EligibilityError(f"forge read {path} returned HTTP {response.status_code}") + try: + return response.json() + except ValueError as exc: + raise EligibilityError(f"forge read {path} returned invalid JSON") from exc + + def verify_commit(self, sha: str) -> None: + body = self._json(f"/git/commits/{sha}") + if not isinstance(body, Mapping) or body.get("sha") != sha: + raise EligibilityError("forge commit does not match scan revision") + + def default_branch_head(self) -> str: + repo = self._json("") + branch = repo.get("default_branch") if isinstance(repo, Mapping) else None + if not isinstance(branch, str) or not branch: + raise EligibilityError("forge repository has no default branch") + body = self._json(f"/branches/{quote(branch, safe='')}") + commit = body.get("commit") if isinstance(body, Mapping) else None + head = commit.get("id") if isinstance(commit, Mapping) else None + if not isinstance(head, str) or not _SHA.fullmatch(head): + raise EligibilityError("forge default-branch head is not a commit SHA") + return head + + def list_markdown(self, directory: str, sha: str) -> list[str] | None: + """Return ``.md`` file paths in ``directory`` at ``sha``; None if absent.""" + response = self._get(f"/contents/{quote(directory)}", ref=sha) + if response.status_code == 404: + return None + if response.status_code != 200: + raise EligibilityError(f"forge listing {directory} returned HTTP {response.status_code}") + entries = response.json() + if not isinstance(entries, list): + raise EligibilityError(f"forge listing {directory} is not a directory") + paths = [] + for entry in entries: + if not isinstance(entry, Mapping): + raise EligibilityError("malformed forge directory entry") + if entry.get("type") == "file" and str(entry.get("name", "")).endswith(".md"): + paths.append(str(entry["path"])) + return sorted(paths) + + def read_text(self, path: str, sha: str) -> str: + response = self._get(f"/raw/{quote(path)}", ref=sha) + if response.status_code != 200: + raise EligibilityError(f"forge raw {path} returned HTTP {response.status_code}") + return response.text + + +# --------------------------------------------------------------------------- # +# Source parsing (mirrors State Hub forge_projection semantics, strictly) +# --------------------------------------------------------------------------- # + + +@dataclass +class _Task: + record_id: str + raw_id: str + workplan_id: str + block: dict[str, Any] + + +@dataclass +class _Workplan: + record_id: str + path: str + meta: dict[str, Any] + tasks: list[_Task] = field(default_factory=list) + + +def _split_frontmatter(text: str, path: str) -> tuple[dict[str, Any], str]: + if not text.startswith("---"): + raise EligibilityError(f"{path}: missing frontmatter") + end = text.find("\n---", 3) + if end == -1: + raise EligibilityError(f"{path}: unterminated frontmatter") + try: + meta = yaml.safe_load(text[3:end]) or {} + except (yaml.YAMLError, ValueError) as exc: + raise EligibilityError(f"{path}: malformed frontmatter") from exc + if not isinstance(meta, dict): + raise EligibilityError(f"{path}: frontmatter is not a mapping") + return meta, text[end + 4 :] + + +def _qualify(task_id: str, workplan_id: str) -> str: + return f"{workplan_id}-{task_id.upper()}" if _SHORT_TASK_RE.match(task_id) else task_id + + +def _dependency_list(value: Any, where: str) -> list[str]: + if value is None or value == "": + return [] + if isinstance(value, str): + items: Sequence[Any] = value.split(",") + elif isinstance(value, list): + if any(isinstance(item, (dict, list)) for item in value): + raise EligibilityError(f"{where}: malformed depends_on") + items = [item for item in value if item is not None] + else: + raise EligibilityError(f"{where}: malformed depends_on") + out: list[str] = [] + for item in items: + target = str(item).strip().strip('"').strip("'").strip() + if target and target not in out: + out.append(target) + return out + + +def _workplan_dependencies(meta: Mapping[str, Any], workplan_id: str) -> list[str]: + out: list[str] = [] + for key in ("depends_on", "depends_on_workplans", "depends_on_tasks"): + for target in _dependency_list(meta.get(key), f"{workplan_id} {key}"): + target = _qualify(target, workplan_id) + if target not in out: + out.append(target) + return out + + +def _parse_workplan(path: str, text: str) -> _Workplan: + meta, body = _split_frontmatter(text, path) + record_id = str(meta.get("id") or "").strip() + if not record_id: + raise EligibilityError(f"{path}: workplan has no id") + workplan = _Workplan(record_id=record_id, path=path, meta=meta) + for match in _TASK_BLOCK_RE.finditer(body): + try: + block = yaml.safe_load(match.group(1).strip()) or {} + except (yaml.YAMLError, ValueError) as exc: + raise EligibilityError(f"{path}: malformed task block") from exc + if not isinstance(block, dict): + continue + raw_id = str(block.get("id") or "").strip() + if not raw_id: + continue + workplan.tasks.append( + _Task(_qualify(raw_id, record_id), raw_id, record_id, block) + ) + return workplan + + +def _status(value: Any, where: str) -> str: + if not isinstance(value, str) or not value.strip(): + raise EligibilityError(f"{where}: missing authoritative status") + return value.strip().lower() + + +def _needs_human(value: Any, where: str) -> bool: + if value is None: + return False # same default as the forge projection + if isinstance(value, bool): + return value + text = str(value).strip().lower() + if text in {"true", "yes", "1", "on"}: + return True + if text in {"false", "no", "0", "off", ""}: + return False + raise EligibilityError(f"{where}: needs_human must be boolean") + + +def _optional_text(value: Any, where: str, name: str) -> str | None: + if value is None: + return None + if not isinstance(value, str): + raise EligibilityError(f"{where}: {name} must be text") + return value.strip() or None + + +def _wait_until(value: Any, where: str) -> str | None: + if value is None or value == "": + return None + if type(value) is date: + return value.isoformat() + text = str(value).strip() + if _DATE.fullmatch(text): + try: + return date.fromisoformat(text).isoformat() + except ValueError: + pass + raise EligibilityError(f"{where}: wait_until must be YYYY-MM-DD") + + +def _flavor(value: Any) -> str | None: + if value is None: + return None + text = str(value).strip().lower() + return None if text in _EMPTY_FLAVOR else text + + +# --------------------------------------------------------------------------- # +# Builder +# --------------------------------------------------------------------------- # + + +def _load_revision(reader: ForgeCommitReader, sha: str) -> list[_Workplan]: + paths: list[str] = [] + for directory in WORKPLAN_DIRS: + listed = reader.list_markdown(directory, sha) + if listed is None: + if directory == WORKPLAN_DIRS[0]: + raise EligibilityError("no workplans directory at scan revision") + continue + paths.extend(listed) + return [_parse_workplan(path, reader.read_text(path, sha)) for path in paths] + + +def build_current_tasks( + *, + target_repo: str, + scan_revision: str, + task_record_ids: Sequence[str], + client: httpx.Client, + decision_status: StatusResolver, + external_status: StatusResolver | None = None, + require_current_head: bool = True, + report: Mapping[str, Any] | None = None, + forge_owner: str = DEFAULT_FORGE_OWNER, + forge_api: str = DEFAULT_FORGE_API, +) -> dict[str, Any]: + """Return ``{"schema", "target_repo", "source_revision", "current_tasks"}``. + + ``current_tasks`` is keyed by the task record IDs requested (as the C-25 + findings name them) and is the exact ``current_tasks`` argument for + ``plan_owner_returns``. Raises :class:`EligibilityError` instead of + returning a partial or unbound mapping. + """ + if not isinstance(target_repo, str) or not _REPO.fullmatch(target_repo): + raise EligibilityError("invalid repository identity") + if not isinstance(scan_revision, str) or not _SHA.fullmatch(scan_revision): + raise EligibilityError("scan_revision must be a full 40-hex commit SHA") + if report is not None: + if report.get("repo_slug") != target_repo: + raise EligibilityError("C-25 report is for a different repository") + for key in ("source_revision", "commit"): + if key in report and report[key] != scan_revision: + raise EligibilityError("C-25 report revision does not match scan_revision") + requested = list(dict.fromkeys(task_record_ids)) + if not requested or any(not isinstance(t, str) or not t for t in requested): + raise EligibilityError("task_record_ids must be non-empty identities") + + reader = ForgeCommitReader(client, target_repo, owner=forge_owner, api_base=forge_api) + reader.verify_commit(scan_revision) + if require_current_head and reader.default_branch_head() != scan_revision: + raise EligibilityError("scan revision is no longer the default-branch head") + + workplans = _load_revision(reader, scan_revision) + by_workplan: dict[str, _Workplan] = {} + by_task: dict[str, _Task] = {} + by_raw: dict[str, list[_Task]] = {} + for workplan in workplans: + if workplan.record_id in by_workplan: + raise EligibilityError(f"duplicate workplan identity {workplan.record_id}") + by_workplan[workplan.record_id] = workplan + for task in workplan.tasks: + if task.record_id in by_task: + raise EligibilityError(f"duplicate task identity {task.record_id}") + by_task[task.record_id] = task + by_raw.setdefault(task.raw_id, []).append(task) + + def predecessor_status(target: str) -> str | None: + if target in by_task: + return _status(by_task[target].block.get("status"), target) + if target in by_workplan: + return _status(by_workplan[target].meta.get("status"), target) + if external_status is None: + return None + try: + value = external_status(target) + except Exception: # noqa: BLE001 - unknown, never guessed + return None + return value.strip().lower() if isinstance(value, str) and value.strip() else None + + rows: dict[str, Any] = {} + for key in requested: + matches = [by_task[key]] if key in by_task else by_raw.get(key, []) + if len(matches) != 1: + raise EligibilityError( + f"task {key} is {'ambiguous' if matches else 'absent'} at scan revision" + ) + task = matches[0] + workplan = by_workplan[task.workplan_id] + where = f"{task.record_id}" + block = task.block + depends_on = _workplan_dependencies(workplan.meta, workplan.record_id) + for target in _dependency_list(block.get("depends_on"), f"{where} depends_on"): + target = _qualify(target, workplan.record_id) + if target not in depends_on: + depends_on.append(target) + decision_id = _optional_text(block.get("decision_id"), where, "decision_id") + resolved_decision: str | None = None + if decision_id is not None: + try: + value = decision_status(decision_id) + except Exception: # noqa: BLE001 - unknown stays unknown + value = None + resolved_decision = value if isinstance(value, str) and value else None + rows[key] = { + "target_repo": target_repo, + "task_record_id": key, + "status": _status(block.get("status"), where), + "workplan_status": _status(workplan.meta.get("status"), workplan.record_id), + "workplan_flavor": _flavor(workplan.meta.get("flavor")), + "needs_human": _needs_human(block.get("needs_human"), where), + "blocked_on": _optional_text(block.get("blocked_on"), where, "blocked_on"), + "depends_on": depends_on, + "predecessor_statuses": {t: predecessor_status(t) for t in depends_on}, + "wait_until": _wait_until(block.get("wait_until"), where), + "decision_id": decision_id, + "decision_status": resolved_decision, + } + return { + "schema": SCHEMA, + "target_repo": target_repo, + "source_revision": scan_revision, + "current_tasks": rows, + } + + +def state_hub_decision_status(client: httpx.Client, api_base: str) -> StatusResolver: + """Resolver reading ``GET {api_base}/decisions/{id}``; failure -> unknown.""" + + def resolve(decision_id: str) -> str | None: + try: + response = client.get(f"{api_base.rstrip('/')}/decisions/{quote(decision_id, safe='')}") + except httpx.HTTPError: + return None + if response.status_code != 200: + return None + try: + status = response.json().get("status") + except (ValueError, AttributeError): + return None + return status if isinstance(status, str) and status else None + + return resolve + + +def forge_head_status_resolver( + client: httpx.Client, + repo_for_prefix: Mapping[str, str], + *, + forge_owner: str = DEFAULT_FORGE_OWNER, + forge_api: str = DEFAULT_FORGE_API, +) -> StatusResolver: + """Cross-repo predecessor resolver over explicit ID-prefix -> repo mapping. + + Each foreign repository is read once, pinned to its default-branch HEAD + commit at first use. Unmapped prefixes and any read failure -> unknown. + """ + cache: dict[str, dict[str, str] | None] = {} + + def load(repo: str) -> dict[str, str] | None: + if repo not in cache: + try: + reader = ForgeCommitReader(client, repo, owner=forge_owner, api_base=forge_api) + head = reader.default_branch_head() + statuses: dict[str, str] = {} + for workplan in _load_revision(reader, head): + statuses[workplan.record_id] = _status(workplan.meta.get("status"), workplan.record_id) + for task in workplan.tasks: + statuses[task.record_id] = _status(task.block.get("status"), task.record_id) + cache[repo] = statuses + except EligibilityError: + cache[repo] = None + return cache[repo] + + def resolve(target: str) -> str | None: + prefix = next( + (p for p in sorted(repo_for_prefix, key=len, reverse=True) if target.startswith(p)), + None, + ) + if prefix is None: + return None + statuses = load(repo_for_prefix[prefix]) + return None if statuses is None else statuses.get(target) + + return resolve diff --git a/tests/test_bounded_operations.py b/tests/test_bounded_operations.py index bbdf003..b09ce87 100644 --- a/tests/test_bounded_operations.py +++ b/tests/test_bounded_operations.py @@ -14,11 +14,13 @@ def _evidence_sinks(event_type: str) -> list[dict[str, str]]: return [{"type": "state-hub-progress", "event_type": event_type}] -def test_registry_names_only_the_three_accepted_operations() -> None: +def test_registry_names_only_the_accepted_operations() -> None: assert set(BOUNDED_OPERATION_REGISTRY) == { "sbom_nexus_ingest", "forgejo_package_prune", "cnpg_option_a_backup", + # ACTIVITY-WP-0044 / RMGR-WP-0020-T03 (draft definition, disabled) + "repo_manager_freshness", } diff --git a/tests/test_owner_return_eligibility.py b/tests/test_owner_return_eligibility.py new file mode 100644 index 0000000..ed875e8 --- /dev/null +++ b/tests/test_owner_return_eligibility.py @@ -0,0 +1,309 @@ +"""Tests for the owner-return eligibility input builder (ACTIVITY-WP-0043-T05). + +All Forge and State Hub HTTP is served by ``httpx.MockTransport``. +""" + +from __future__ import annotations + +import importlib.util +from datetime import date +from pathlib import Path +from urllib.parse import unquote + +import httpx +import pytest + +from activity_core import owner_return_eligibility as ore +from activity_core import owner_return_pickup as orp + +REPO = "ops-warden" +SHA = "a" * 40 +OLD = "b" * 40 +API = ore.DEFAULT_FORGE_API +BASE = f"{API}/repos/coulomb/{REPO}" + +WP1 = """--- +id: WARDEN-WP-0041 +status: active +flavor: implementation +depends_on: [WARDEN-WP-0040] +--- + +## Owner return + +```task +id: WARDEN-WP-0041-T03 +status: wait +priority: high +needs_human: false +blocked_on: "message-from:state-hub" +depends_on: [T01, STATE-WP-0094-T03] +wait_until: "2026-10-01" +decision_id: "dec-1" +``` + +```task +id: T01 +status: done +``` + +```task +id: WARDEN-WP-0041-T04 +status: wait +blocked_on: "message-from:state-hub" +``` +""" + +WP0 = """--- +id: WARDEN-WP-0040 +status: finished +--- +""" + + +def _files(**over): + files = { + "workplans/WARDEN-WP-0041-owner.md": WP1, + "workplans/archived/261001-WARDEN-WP-0040-prior.md": WP0, + } + files.update(over) + return {k: v for k, v in files.items() if v is not None} + + +def _transport(files, *, sha=SHA, head=SHA, seen=None, fail=None): + def handler(request: httpx.Request) -> httpx.Response: + url = str(request.url).split("?")[0] + path = unquote(url) + ref = request.url.params.get("ref") + if seen is not None: + seen.append((path, ref)) + if fail and fail in path: + return httpx.Response(500) + if path == BASE: + return httpx.Response(200, json={"default_branch": "main"}) + if path == f"{BASE}/branches/main": + return httpx.Response(200, json={"commit": {"id": head}}) + if path.startswith(f"{BASE}/git/commits/"): + asked = path.rsplit("/", 1)[1] + return httpx.Response(200, json={"sha": sha if asked == sha else "c" * 40}) + if path.startswith(f"{BASE}/contents/"): + if ref != sha: + return httpx.Response(404) + directory = path[len(f"{BASE}/contents/"):] + entries = [ + {"type": "file", "name": p.rsplit("/", 1)[1], "path": p} + for p in files if p.rsplit("/", 1)[0] == directory + ] + if not entries and not any(p.startswith(directory + "/") for p in files): + return httpx.Response(404) + if directory == "workplans": + entries.append({"type": "dir", "name": "archived", "path": "workplans/archived"}) + return httpx.Response(200, json=entries) + if path.startswith(f"{BASE}/raw/"): + name = path[len(f"{BASE}/raw/"):] + if ref != sha or name not in files: + return httpx.Response(404) + return httpx.Response(200, text=files[name]) + return httpx.Response(404) + + return httpx.MockTransport(handler) + + +def _build(files=None, *, decisions=None, external=None, **kw): + transport = kw.pop("transport", None) or _transport(files or _files()) + client = httpx.Client(transport=transport) + decisions = {"dec-1": "resolved"} if decisions is None else decisions + return ore.build_current_tasks( + target_repo=REPO, + scan_revision=kw.pop("scan_revision", SHA), + task_record_ids=kw.pop("task_record_ids", ["WARDEN-WP-0041-T03"]), + client=client, + decision_status=decisions.get, + external_status=external, + **kw, + ) + + +def test_builds_complete_row_bound_to_revision(): + seen: list = [] + out = _build(transport=_transport(_files(), seen=seen), + external={"STATE-WP-0094-T03": "done"}.get) + assert out["schema"] == ore.SCHEMA + assert out["source_revision"] == SHA + row = out["current_tasks"]["WARDEN-WP-0041-T03"] + assert row == { + "target_repo": REPO, + "task_record_id": "WARDEN-WP-0041-T03", + "status": "wait", + "workplan_status": "active", + "workplan_flavor": "implementation", + "needs_human": False, + "blocked_on": "message-from:state-hub", + "depends_on": ["WARDEN-WP-0040", "WARDEN-WP-0041-T01", "STATE-WP-0094-T03"], + "predecessor_statuses": { + "WARDEN-WP-0040": "finished", + "WARDEN-WP-0041-T01": "done", + "STATE-WP-0094-T03": "done", + }, + "wait_until": "2026-10-01", + "decision_id": "dec-1", + "decision_status": "resolved", + } + content_reads = [(p, r) for p, r in seen if "/contents/" in p or "/raw/" in p] + assert content_reads and all(ref == SHA for _, ref in content_reads) + + +def test_unresolvable_predecessor_and_decision_stay_unknown(): + out = _build(decisions={}) + row = out["current_tasks"]["WARDEN-WP-0041-T03"] + assert row["predecessor_statuses"]["STATE-WP-0094-T03"] is None + assert "STATE-WP-0094-T03" in row["depends_on"] + assert row["decision_status"] is None + + +def test_external_resolver_exception_is_unknown_not_fatal(): + def boom(_): + raise RuntimeError("down") + row = _build(external=boom)["current_tasks"]["WARDEN-WP-0041-T03"] + assert row["predecessor_statuses"]["STATE-WP-0094-T03"] is None + + +def test_task_without_dependencies_has_explicit_workplan_gate_only(): + row = _build(task_record_ids=["WARDEN-WP-0041-T04"])["current_tasks"]["WARDEN-WP-0041-T04"] + assert row["depends_on"] == ["WARDEN-WP-0040"] + assert row["decision_id"] is None and row["wait_until"] is None + assert row["needs_human"] is False + + +@pytest.mark.parametrize("revision", ["main", "a" * 39, "A" * 40, None]) +def test_refuses_non_sha_revision(revision): + with pytest.raises(ore.EligibilityError): + _build(scan_revision=revision) + + +def test_refuses_when_scan_is_not_current_head(): + with pytest.raises(ore.EligibilityError, match="no longer the default-branch head"): + _build(transport=_transport(_files(), head=OLD)) + + +def test_stale_scan_allowed_only_when_head_check_disabled(): + out = _build(transport=_transport(_files(), head=OLD), require_current_head=False) + assert out["source_revision"] == SHA + + +def test_refuses_unknown_commit(): + with pytest.raises(ore.EligibilityError, match="does not match"): + _build(transport=_transport(_files(), sha=OLD, head=SHA)) + + +def test_refuses_report_revision_mismatch(): + report = {"repo_slug": REPO, "source_revision": OLD, "issues": []} + with pytest.raises(ore.EligibilityError, match="revision"): + _build(report=report) + with pytest.raises(ore.EligibilityError, match="different repository"): + _build(report={"repo_slug": "other", "issues": []}) + + +@pytest.mark.parametrize( + "old,new,match", + [ + ("status: wait\npriority: high", "priority: high", "missing authoritative status"), + ("needs_human: false", "needs_human: maybe", "needs_human"), + ('wait_until: "2026-10-01"', 'wait_until: "soon"', "wait_until"), + ("depends_on: [T01, STATE-WP-0094-T03]", "depends_on: {a: b}", "depends_on"), + ("id: WARDEN-WP-0041\nstatus: active", "id: WARDEN-WP-0041", "missing authoritative status"), + ], +) +def test_missing_or_malformed_fields_fail_closed(old, new, match): + files = _files(**{"workplans/WARDEN-WP-0041-owner.md": WP1.replace(old, new, 1)}) + with pytest.raises(ore.EligibilityError, match=match): + _build(files) + + +def test_absent_and_duplicate_tasks_refuse(): + with pytest.raises(ore.EligibilityError, match="absent"): + _build(task_record_ids=["WARDEN-WP-0041-T09"]) + dup = WP1.replace("id: WARDEN-WP-0041-T04", "id: WARDEN-WP-0041-T03") + with pytest.raises(ore.EligibilityError, match="duplicate task"): + _build(_files(**{"workplans/WARDEN-WP-0041-owner.md": dup})) + + +def test_forge_failure_refuses(): + with pytest.raises(ore.EligibilityError): + _build(transport=_transport(_files(), fail="/raw/")) + + +def test_missing_workplans_directory_refuses(): + with pytest.raises(ore.EligibilityError, match="no workplans"): + _build(transport=_transport({})) + + +def test_state_hub_decision_resolver(): + def handler(request): + if request.url.path.endswith("/decisions/dec-1"): + return httpx.Response(200, json={"id": "dec-1", "status": "open"}) + return httpx.Response(404) + resolve = ore.state_hub_decision_status( + httpx.Client(transport=httpx.MockTransport(handler)), "http://hub" + ) + assert resolve("dec-1") == "open" + assert resolve("dec-2") is None + + +def test_forge_head_resolver_pins_foreign_head_and_maps_prefix(): + foreign = "d" * 40 + files = {"workplans/STATE-WP-0094-x.md": "---\nid: STATE-WP-0094\nstatus: active\n---\n" + "```task\nid: STATE-WP-0094-T03\nstatus: progress\n```\n"} + fbase = f"{API}/repos/coulomb/state-hub" + + def handler(request): + path = unquote(str(request.url).split("?")[0]) + ref = request.url.params.get("ref") + if path == fbase: + return httpx.Response(200, json={"default_branch": "main"}) + if path == f"{fbase}/branches/main": + return httpx.Response(200, json={"commit": {"id": foreign}}) + if path == f"{fbase}/contents/workplans" and ref == foreign: + return httpx.Response(200, json=[{"type": "file", "name": "STATE-WP-0094-x.md", + "path": "workplans/STATE-WP-0094-x.md"}]) + if path.startswith(f"{fbase}/raw/") and ref == foreign: + return httpx.Response(200, text=files[path[len(f"{fbase}/raw/"):]]) + return httpx.Response(404) + + resolve = ore.forge_head_status_resolver( + httpx.Client(transport=httpx.MockTransport(handler)), {"STATE-WP-": "state-hub"} + ) + assert resolve("STATE-WP-0094-T03") == "progress" + assert resolve("STATE-WP-0094") == "active" + assert resolve("STATE-WP-0094-T99") is None + assert resolve("OTHER-WP-0001") is None + + +_PLANNER = Path("/home/worsch/state-hub/scripts/owner_return_handoff.py") + + +@pytest.mark.skipif(not _PLANNER.is_file(), reason="state-hub planner not checked out") +def test_output_is_accepted_by_state_hub_planner(): + spec = importlib.util.spec_from_file_location("owner_return_handoff", _PLANNER) + planner = importlib.util.module_from_spec(spec) + spec.loader.exec_module(planner) + msg = "11111111-2222-3333-4444-555555555555" + report = {"repo_slug": REPO, "result": "warn", "issues": [{ + "check_id": "C-25", "file_value": "message-from:state-hub", + "owner_return": {"task_record_id": "WARDEN-WP-0041-T03", "message_ids": [msg]}, + }]} + ok = _build(external={"STATE-WP-0094-T03": "done"}.get, report=report) + plan = planner.plan_owner_returns( + report, activity_definition_id=orp.ACTIVITY_DEFINITION_ID, + current_tasks=ok["current_tasks"], observed_on=date(2026, 10, 4), + ) + assert [c["disposition"] for c in plan] == ["reviewable"] + assert plan[0]["triggering_event_id"] == orp.owner_return_event_id( + REPO, "WARDEN-WP-0041-T03", msg) + unknown = _build(report=report) + plan = planner.plan_owner_returns( + report, activity_definition_id=orp.ACTIVITY_DEFINITION_ID, + current_tasks=unknown["current_tasks"], observed_on=date(2026, 10, 4), + ) + assert plan[0]["disposition"] == "stopped" + assert plan[0]["stop_reason"] == "predecessor_not_terminal_or_unknown" diff --git a/tests/test_repo_manager_freshness_definition.py b/tests/test_repo_manager_freshness_definition.py new file mode 100644 index 0000000..65551fd --- /dev/null +++ b/tests/test_repo_manager_freshness_definition.py @@ -0,0 +1,170 @@ +"""Draft repo-manager-bounded-freshness definition (ACTIVITY-WP-0044 / RMGR-WP-0020-T03).""" + +from __future__ import annotations + +import json +import subprocess +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +from sqlalchemy.dialects import postgresql + +from activity_core import activities, sync_activity_definitions, sync_schedules +from activity_core.bounded_operations import ( + BOUNDED_OPERATION_REGISTRY, + normalize_operation_result, + operation_sources, + validate_bounded_operations, +) +from activity_core.context_resolvers import repo_manager +from activity_core.definition_parser import parse_file + +DEF_PATH = ( + Path(__file__).parent.parent / "activity-definitions" / "repo-manager-bounded-freshness.md" +) + + +def _source(**params): + base = {"limit": 6, "max_seconds": 240, "apply": True} + base.update(params) + return { + "type": "repo-manager", + "query": "freshness_run", + "operation": "repo_manager_freshness", + "required": True, + "params": base, + "bind_to": "context.freshness", + } + + +def test_definition_parses_disabled_with_reviewed_shape(): + d = parse_file(DEF_PATH) + assert d.id == "repo-manager-bounded-freshness" + assert d.enabled is False + tc = d.trigger_config + assert tc["trigger_type"] == "cron" + assert tc["cron_expression"] == "40 */2 * * *" + assert tc["timezone"] == "UTC" + assert tc["misfire_policy"] == "skip" + [(source, spec)] = operation_sources(d.context_sources) + assert spec.operation_id == "repo_manager_freshness" + assert source["params"] == {"limit": 6, "max_seconds": 240, "apply": True} + assert all(rule["condition"] == "False" for rule in d.rules) + + +def test_operation_is_registered_with_single_attempt_and_timeout(): + spec = BOUNDED_OPERATION_REGISTRY["repo_manager_freshness"] + assert spec.temporal_max_attempts == 1 + assert spec.resolve_before_execute is False + assert spec.max_timeout_seconds >= 240 + repo_manager._TIMEOUT_GRACE_SECONDS + + +@pytest.mark.parametrize( + "params,error", + [ + ({"limit": 0}, "limit"), + ({"limit": 13}, "limit"), + ({"limit": True}, "limit"), + ({"max_seconds": 271}, "max_seconds"), + ({"max_seconds": "240"}, "max_seconds"), + ({"apply": "yes"}, "apply"), + ({"command": "rm -rf /"}, "code-fixed"), + ({"rmgr_bin": "/tmp/x"}, "code-fixed"), + ], +) +def test_admission_fails_closed(params, error): + with pytest.raises(ValueError, match=error): + validate_bounded_operations([_source(**params)], []) + + +def test_argv_is_code_fixed(monkeypatch): + monkeypatch.delenv("ACTIVITY_CORE_RMGR_BIN", raising=False) + assert repo_manager.build_freshness_argv({"limit": 6, "max_seconds": 240, "apply": True}) == [ + "rmgr", "freshness", "run", "--limit", "6", "--max-seconds", "240", + ] + assert repo_manager.build_freshness_argv( + {"limit": 3, "max_seconds": 60, "apply": False} + )[-1] == "--dry-run" + + +def test_invocation_projects_counts_only(monkeypatch): + calls = [] + + def fake_run(argv, **kwargs): + calls.append((argv, kwargs["timeout"])) + out = {"receipts": [ + {"repo": "a", "outcome": "applied", "detail": "must not persist"}, + {"repo": "b", "outcome": "noop"}, + {"repo": "c", "outcome": "refused"}, + {"repo": "d", "outcome": "bogus"}, + ]} + return subprocess.CompletedProcess(argv, 0, json.dumps(out), "") + + monkeypatch.setattr(repo_manager.subprocess, "run", fake_run) + raw = repo_manager.repo_manager_freshness({"limit": 6, "max_seconds": 240, "apply": True}) + projected = normalize_operation_result("repo_manager_freshness", raw) + assert calls[0][1] == 240 + repo_manager._TIMEOUT_GRACE_SECONDS + assert projected["receipt_count"] == 4 + assert projected["outcomes"]["applied"] == 1 + assert projected["outcomes"]["noop"] == 1 + assert projected["outcomes"]["refused"] == 1 + assert projected["apply"] is True and projected["exit_code"] == 0 + assert "must not persist" not in json.dumps(projected) + + +def test_invocation_fails_on_non_json(monkeypatch): + monkeypatch.setattr( + repo_manager.subprocess, "run", + lambda argv, **kw: subprocess.CompletedProcess(argv, 2, "", "boom"), + ) + with pytest.raises(RuntimeError, match="exit 2"): + repo_manager.repo_manager_freshness({"limit": 1, "max_seconds": 10, "apply": False}) + + +@pytest.mark.asyncio +async def test_execute_bounded_operation_dispatches_to_repo_manager(monkeypatch): + monkeypatch.setattr( + repo_manager, "repo_manager_freshness", + lambda params: {"receipts": [], "apply": False, "limit": 1, "max_seconds": 10}, + ) + patch = await activities.execute_bounded_operation({"source": _source(apply=False)}) + assert patch["freshness"]["kind"] == "repo_manager_freshness" + + +@pytest.mark.asyncio +async def test_sync_upserts_disabled_row_and_paused_schedule(monkeypatch): + d = parse_file(DEF_PATH) + monkeypatch.setattr(sync_activity_definitions, "scan_and_parse", lambda: [d]) + statements = [] + session = MagicMock() + session.execute = AsyncMock(side_effect=lambda stmt: statements.append(stmt)) + begin = MagicMock() + begin.__aenter__ = AsyncMock(return_value=None) + begin.__aexit__ = AsyncMock(return_value=False) + session.begin = MagicMock(return_value=begin) + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=session) + ctx.__aexit__ = AsyncMock(return_value=False) + + assert await sync_activity_definitions.sync(lambda: ctx) == 1 + params = statements[0].compile(dialect=postgresql.dialect()).params + assert params["enabled"] is False + assert params["id"] == sync_activity_definitions._definition_uuid(d.id) + + upserted = [] + + async def fake_upsert(client, defn): + upserted.append(defn) + + monkeypatch.setattr(sync_schedules, "upsert_schedule", fake_upsert) + monkeypatch.setattr(sync_schedules, "list_schedules", AsyncMock(return_value=[])) + row = SimpleNamespace( + id=params["id"], name=d.name, enabled=params["enabled"], + trigger_config=d.trigger_config, context_sources=d.context_sources, + task_templates=[], dedupe_key_strategy="skip", version=1, + ) + result = await sync_schedules.sync_schedule_rows(object(), [row]) + assert result.upserted == 0 and result.paused == 1 + assert upserted[0].enabled is False diff --git a/workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md b/workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md index 563759b..267575b 100644 --- a/workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md +++ b/workplans/ACTIVITY-WP-0043-state-hub-owner-return-pickup.md @@ -85,5 +85,26 @@ blocking_reason: "Activation is a separately reviewed founder action" state_hub_task_id: "254526ce-5a76-57b3-b7d5-a874518c7b40" ``` -Plan: `docs/owner-return-pickup-activation-plan.md`. Requires T03 and an -eligibility-source binding (fresh authoritative `current_tasks`). +Plan: `docs/owner-return-pickup-activation-plan.md`. Requires T03 and T05 +(eligibility-source binding of fresh authoritative `current_tasks`). + +## Eligibility input builder bound to the scan revision + +```task +id: ACTIVITY-WP-0043-T05 +status: done +priority: high +``` + +`src/activity_core/owner_return_eligibility.py::build_current_tasks` builds the +planner's `current_tasks` from immutable Forge commit content at the exact +40-hex `scan_revision` of the C-25 scan (`/contents` and `/raw` with +`ref=`; no checkouts, ADR-002). It verifies the commit and, by default, +that it is still the default-branch head; refuses on any revision mismatch, +missing/malformed field, duplicate identity or absent task. Predecessors are +task-block plus workplan-frontmatter dependencies, resolved in the same commit +or via an explicit cross-repo resolver (`forge_head_status_resolver`, HEAD +pinned per foreign repo); unresolved targets stay `None`. Decision status comes +from an injected resolver (`state_hub_decision_status`); failure stays unknown. +Tests: `tests/test_owner_return_eligibility.py` (HTTP via MockTransport, +including acceptance by the State Hub planner). diff --git a/workplans/ACTIVITY-WP-0044-repo-manager-bounded-freshness-definition.md b/workplans/ACTIVITY-WP-0044-repo-manager-bounded-freshness-definition.md new file mode 100644 index 0000000..07a8908 --- /dev/null +++ b/workplans/ACTIVITY-WP-0044-repo-manager-bounded-freshness-definition.md @@ -0,0 +1,72 @@ +--- +id: ACTIVITY-WP-0044 +type: workplan +title: "Repo Manager bounded freshness ActivityDefinition (RMGR-WP-0020-T03)" +domain: infotech +repo: activity-core +status: active +owner: claude-activity-core +topic_slug: activity-core +created: "2026-10-04" +updated: "2026-10-04" +flavor: implementation +related: [RMGR-WP-0020, STATE-WP-0094] +--- + +Origin: Repo Manager RMGR-WP-0020-T03 hands Activity Core a draft definition +for the bounded Forge-sourced freshness runner. Repo Manager owns the engine +(`rmgr freshness run`, RMGR-WP-0020-T02) and receipts (T01); Activity Core owns +the schedule. Source level only: no schedule enabled, no deployment, no +credential handling, no run dispatched. + +## Draft definition and bounded operation (disabled) + +```task +id: ACTIVITY-WP-0044-T01 +status: done +priority: medium +``` + +- `activity-definitions/repo-manager-bounded-freshness.md`: `enabled: false`, + cron `40 */2 * * *` UTC, `misfire_policy: skip`, `limit: 6`, + `max_seconds: 240`, `apply: true`; rule condition `False` (no task spawning). +- ACT-ADR-007 registry entry `repo_manager_freshness` (source `repo-manager` / + `freshness_run`): single Temporal attempt, timeout 300 s, admission requires + explicit boolean `apply`, `limit` 1..12, `max_seconds` 1..270, and refuses + any definition-supplied command/binary. +- `context_resolvers/repo_manager.py`: code-fixed argv + `rmgr freshness run --limit N --max-seconds S [--dry-run]` (executable from + worker env `ACTIVITY_CORE_RMGR_BIN`, default `rmgr`); only a counted + projection (receipt count per outcome, exit code) is persisted. +- Tests: `tests/test_repo_manager_freshness_definition.py` (parse, admission, + argv, projection, dispatch, DB sync stays disabled, schedule sync paused). + +## Interface confirmation against the Repo Manager CLI + +```task +id: ACTIVITY-WP-0044-T02 +status: wait +priority: medium +needs_human: false +blocking_reason: "rmgr freshness run is being built in RMGR-WP-0020-T02; JSON output (receipts[].outcome) assumed from the T01 receipt contract" +depends_on: [RMGR-WP-0020-T01, RMGR-WP-0020-T02] +``` + +When RMGR-WP-0020-T01/T02 land, confirm the CLI flags and that stdout is one +JSON object with `receipts[]` carrying `repo-manager.freshness-receipt.v1` +outcomes; adjust `normalize_operation_result` if the shape differs. Decide the +worker image/packaging of `rmgr` (no deployment in this workplan). + +## Founder-reviewed activation + +```task +id: ACTIVITY-WP-0044-T03 +status: wait +priority: medium +needs_human: true +blocking_reason: "Activation follows RMGR-WP-0020-T04 and is a founder action" +depends_on: [RMGR-WP-0020-T04] +``` + +Flip `enabled: true` only after the RMGR-WP-0020-T04 manual dry-run/apply fires. +Rollback: `enabled: false` (or `status: paused`).