Compare commits
2 commits
9741d5cc19
...
ba4ef00128
| Author | SHA1 | Date | |
|---|---|---|---|
| ba4ef00128 | |||
| c6df768ef0 |
13 changed files with 1257 additions and 4 deletions
63
activity-definitions/repo-manager-bounded-freshness.md
Normal file
63
activity-definitions/repo-manager-bounded-freshness.md
Normal file
|
|
@ -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"]
|
||||
```
|
||||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
63
src/activity_core/context_resolvers/repo_manager.py
Normal file
63
src/activity_core/context_resolvers/repo_manager.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
474
src/activity_core/owner_return_eligibility.py
Normal file
474
src/activity_core/owner_return_eligibility.py
Normal file
|
|
@ -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=<sha>``), 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/<sha>`` 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
|
||||
|
|
@ -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",
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
309
tests/test_owner_return_eligibility.py
Normal file
309
tests/test_owner_return_eligibility.py
Normal file
|
|
@ -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"
|
||||
170
tests/test_repo_manager_freshness_definition.py
Normal file
170
tests/test_repo_manager_freshness_definition.py
Normal file
|
|
@ -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
|
||||
|
|
@ -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=<sha>`; 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).
|
||||
|
|
|
|||
|
|
@ -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`).
|
||||
Loading…
Add table
Add a link
Reference in a new issue