From 58414404d60f936277cbfb05d6c07113320fd4dd Mon Sep 17 00:00:00 2001 From: tegwick Date: Sun, 30 Aug 2026 22:38:54 +0200 Subject: [PATCH] feat: add governed fast work-record sync Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c --- AGENTS.md | 29 +- README.md | 11 + docs/repository-standards_v0.1.md | 21 +- src/repo_manager/cli.py | 133 ++++++ src/repo_manager/commands/task.py | 424 ++++++++++++++++++ src/repo_manager/commands/workplan.py | 26 +- src/repo_manager/identifiers.py | 107 ++++- src/repo_manager/projection_sync.py | 239 ++++++++++ tests/test_fast_work_records.py | 236 ++++++++++ ...ast-work-record-mutation-and-projection.md | 86 ++++ 10 files changed, 1280 insertions(+), 32 deletions(-) create mode 100644 src/repo_manager/commands/task.py create mode 100644 src/repo_manager/projection_sync.py create mode 100644 tests/test_fast_work_records.py create mode 100644 workplans/RMGR-WP-0012-fast-work-record-mutation-and-projection.md diff --git a/AGENTS.md b/AGENTS.md index e4ef0da..5d3ddc1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -120,20 +120,13 @@ curl -s -X PATCH "http://127.0.0.1:8000/tasks/" \ 3. Log: `POST /progress/` with a summary of what changed (name handoff ids) 4. After workplan file changes, run: ```bash - statehub fix-consistency + rmgr sync --path . --push ``` - Coding agents should run this directly; ask the operator only if the CLI or - State Hub API is unavailable. This syncs task status from files into the hub DB. - If C-06/C-11 says this host is not the identifier registrar, do not retry - and do not set `STATEHUB_REGISTRAR` yourself. Use the governed, single-repo - fallback once the branch is clean and synchronized: - ```bash - uv run --project ~/repo-manager rmgr registrar-reconcile \ - --path . --confirm-primary --push - ``` - If that command is unavailable, send one registrar request to `repo-manager` - naming the repository and missing canonical ids, then continue file-backed - work without UUIDs. + The command assigns only missing deterministic identifiers, verifies the + pushed Forgejo commit and `primary/railliance01`, then requests one central + forge-derived reconciliation. A queued receipt is pending evidence; rerun + after connectivity returns. Use `statehub fix-consistency` separately for a + deep audit. `registrar-reconcile` is legacy migration/repair only. --- @@ -182,7 +175,7 @@ owner: codex topic_slug: ... created: "YYYY-MM-DD" updated: "YYYY-MM-DD" -state_hub_workstream_id: "" # fix-consistency — do not edit (legacy field name; workplan UUID) +state_hub_workstream_id: "" # deterministic UUIDv5; managed by Repo Manager --- ``` @@ -203,7 +196,7 @@ API/MCP/frontmatter bridges until `STATE-WP-0069` retires them — see id: RMGR-WP-NNNN-T01 status: wait | todo | progress | done | cancel priority: high | medium | low -state_hub_task_id: "" # written by fix-consistency — do not edit +state_hub_task_id: "" # deterministic UUIDv5; managed by Repo Manager ` ` ` Task description text. @@ -218,7 +211,5 @@ not a kind. Fleet list lives on State Hub, not in `SCOPE.md`. To create a new workplan: 1. Write the file following the format above -2. Run `statehub fix-consistency` locally. -3. On a non-registrar C-06/C-11 skip, invoke `rmgr registrar-reconcile` once as - documented in the session-close protocol. Never set the registrar environment - variable directly and never send repeated requests for the same ids. +2. Run `rmgr sync --path . --push`. +3. Run `statehub fix-consistency` only when a separate deep audit is needed. diff --git a/README.md b/README.md index dd0906e..06a0af5 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,10 @@ rmgr update-task-status --path . --task-id --status progress rmgr workplan create --path . --workplan-id EX-WP-0001 --title "Example" --goal "Deliver X." rmgr workplan update --path . --workplan-id EX-WP-0001 --status active rmgr workplan delete --path . --workplan-id EX-WP-0001 --confirm # archives; never erases +rmgr task add --path . --workplan-id EX-WP-0001 --title "Implement" --description "Deliver it." +rmgr task update --path . --task-id EX-WP-0001-T01 --status progress +rmgr adhoc add --path . --title "Small fix" --description "Complete the bounded fix." +rmgr sync --path . --push # primary/railliance01 derives the exact pushed commit rmgr register put --path . --kind technical-debt --entry-id TD-001 \ --title "Debt item" --data-json '{"severity":"high"}' rmgr register list --path . --kind technical-debt @@ -51,6 +55,13 @@ rmgr scaffold --path ../prj-example --flavor project --wp-prefix EX-WP --no-comm rmgr scaffold --path . --refresh-hub-access --no-commit # State Hub access table from config/state-hub-access.yaml ``` +`rmgr sync` is the normal work-record closeout path. It uses the canonical kind +registry, fills only missing deterministic UUIDv5 identifiers, refuses dirty or +behind workplan sources, verifies that the API is `primary/railliance01`, and +requests one central forge-derived reconciliation. A local/cache State Hub is +never populated. `registrar-reconcile` remains only for sealed legacy UUID +migration and repair. + Vertical-slice proof: [docs/evidence/t05-vertical-slice.md](docs/evidence/t05-vertical-slice.md). Direction: diff --git a/docs/repository-standards_v0.1.md b/docs/repository-standards_v0.1.md index d3721df..e96c8c2 100644 --- a/docs/repository-standards_v0.1.md +++ b/docs/repository-standards_v0.1.md @@ -48,14 +48,21 @@ not write new hub primary keys into git. | env unset, hostname starts with `railiance` | yes | | env unset, any other hostname | no | -Accepted cost: new workplan/task registration requires connectivity to -the registrar. Disconnected work cannot register until T03. Implementation: -`repo_manager.registrar.is_identifier_registrar`; consumed by -`statehub fix-consistency` C-06 / C-11 / C-32. +The interim registrar rule is superseded for new canonical workplans and tasks. +Any host may derive the same missing UUIDv5 value; existing UUIDs are preserved +and replacement remains a separately sealed migration. The normal path is: -The production fleet sweep is disabled and is not the interactive recovery -path. Repo Manager provides a bounded on-demand registrar for one clean, -up-to-date repository at a time: +```bash +rmgr sync --path . --push +``` + +This verifies that repository sources are committed and visible on the forge, +verifies the State Hub identity, and requests one central reconciliation of the +exact pushed commit. Disconnected work remains valid in files and receives an +explicit pending receipt rather than being written to a local cache database. + +The production fleet sweep remains disabled. The bounded on-demand registrar +below is retained only for sealed legacy identifier migration and repair: ```bash uv run --project ~/repo-manager rmgr registrar-reconcile \ diff --git a/src/repo_manager/cli.py b/src/repo_manager/cli.py index c9ed27a..524ce16 100644 --- a/src/repo_manager/cli.py +++ b/src/repo_manager/cli.py @@ -6,6 +6,7 @@ import argparse import json import os import sys +from datetime import date from pathlib import Path from repo_manager.commands.rapp import add_rapp_parser @@ -129,6 +130,67 @@ def main(argv: list[str] | None = None) -> int: help="Patch file only (invalid as full applied evidence; for tests)", ) + p_task = sub.add_parser("task", help="Governed file-backed task mutations") + task_sub = p_task.add_subparsers(dest="task_command") + p_task_add = task_sub.add_parser("add", help="Append a task to a workplan") + p_task_add.add_argument("--path", default=".") + p_task_add.add_argument("--workplan-id", required=True) + p_task_add.add_argument("--task-id", default=None) + p_task_add.add_argument("--title", required=True) + p_task_add.add_argument("--description", required=True) + p_task_add.add_argument("--status", choices=["wait", "todo", "progress", "done", "cancel"], default="todo") + p_task_add.add_argument("--priority", choices=["low", "medium", "high", "critical"], default="medium") + p_task_add.add_argument("--needs-human", action="store_true") + p_task_add.add_argument("--intervention-note", default=None) + p_task_add.add_argument("--blocking-reason", default=None) + p_task_update = task_sub.add_parser("update", help="Update task metadata and lifecycle") + p_task_update.add_argument("--path", default=".") + p_task_update.add_argument("--task-id", required=True) + p_task_update.add_argument("--title", default=None) + p_task_update.add_argument("--description", default=None) + p_task_update.add_argument("--status", choices=["wait", "todo", "progress", "done", "cancel"], default=None) + p_task_update.add_argument("--priority", choices=["low", "medium", "high", "critical"], default=None) + p_task_update.add_argument("--needs-human", action=argparse.BooleanOptionalAction, default=None) + p_task_update.add_argument("--intervention-note", default=None) + p_task_update.add_argument("--blocking-reason", default=None) + for task_parser in (p_task_add, p_task_update): + task_parser.add_argument("--reason", default="rmgr CLI") + task_parser.add_argument("--correlation-id", default=None) + task_parser.add_argument("--idempotency-key", default=None) + task_parser.add_argument("--expected-head-sha", default=None) + task_parser.add_argument("--slug", default=None) + task_parser.add_argument("--push", action="store_true") + task_parser.add_argument("--no-commit", action="store_true") + + p_adhoc = sub.add_parser("adhoc", help="Daily repository-qualified ad-hoc work") + adhoc_sub = p_adhoc.add_subparsers(dest="adhoc_command") + p_adhoc_add = adhoc_sub.add_parser("add", help="Create/reuse today's ad-hoc and append a task") + p_adhoc_add.add_argument("--path", default=".") + p_adhoc_add.add_argument("--title", required=True) + p_adhoc_add.add_argument("--description", required=True) + p_adhoc_add.add_argument("--prefix", default=None) + p_adhoc_add.add_argument("--date", type=date.fromisoformat, default=None) + p_adhoc_add.add_argument("--status", choices=["wait", "todo", "progress", "done", "cancel"], default="todo") + p_adhoc_add.add_argument("--priority", choices=["low", "medium", "high", "critical"], default="medium") + p_adhoc_add.add_argument("--owner", default="codex") + p_adhoc_add.add_argument("--reason", default="rmgr CLI") + p_adhoc_add.add_argument("--correlation-id", default=None) + p_adhoc_add.add_argument("--idempotency-key", default=None) + p_adhoc_add.add_argument("--expected-head-sha", default=None) + p_adhoc_add.add_argument("--slug", default=None) + p_adhoc_add.add_argument("--push", action="store_true") + p_adhoc_add.add_argument("--no-commit", action="store_true") + + p_sync = sub.add_parser("sync", help="Fast authoritative forge projection reconciliation") + p_sync.add_argument("--path", default=".") + p_sync.add_argument("--slug", default=None) + p_sync.add_argument("--api-base", default=os.environ.get("STATE_HUB_API_BASE", "http://127.0.0.1:8000")) + p_sync.add_argument("--expected-instance-label", default="railliance01") + p_sync.add_argument("--push", action="store_true", help="Push ahead commits before reconciliation") + p_sync.add_argument("--no-commit-identifiers", action="store_true") + p_sync.add_argument("--acknowledge-retirements", action="store_true") + p_sync.add_argument("--no-queue", action="store_true") + p_wp = sub.add_parser("workplan", help="Governed file-backed workplan mutations") wp_sub = p_wp.add_subparsers(dest="workplan_command") p_wp_create = wp_sub.add_parser("create", help="Create a workplan file") @@ -581,6 +643,77 @@ def main(argv: list[str] | None = None) -> int: print(json.dumps(result.to_dict(), indent=2)) return 0 if result.status == "applied" else 1 + if args.command == "task": + if not args.task_command: + p_task.print_help() + return 2 + from repo_manager.commands.task import mutate_task + + result = mutate_task( + Path(args.path), + args.task_id, + operation=args.task_command, + workplan_id=getattr(args, "workplan_id", None), + title=args.title, + description=args.description, + status=args.status, + priority=args.priority, + needs_human=args.needs_human, + intervention_note=args.intervention_note, + blocking_reason=args.blocking_reason, + correlation_id=args.correlation_id, + reason=args.reason, + commit=not args.no_commit, + push=args.push, + expected_head_sha=args.expected_head_sha, + idempotency_key=args.idempotency_key, + repo_slug=args.slug, + ) + print(json.dumps(result.to_dict(), indent=2)) + return 0 if result.status == "applied" else 1 + + if args.command == "adhoc": + if args.adhoc_command != "add": + p_adhoc.print_help() + return 2 + from repo_manager.commands.task import add_adhoc_task + + result = add_adhoc_task( + Path(args.path), + args.title, + args.description, + prefix=args.prefix, + on_date=args.date, + status=args.status, + priority=args.priority, + owner=args.owner, + correlation_id=args.correlation_id, + reason=args.reason, + commit=not args.no_commit, + push=args.push, + expected_head_sha=args.expected_head_sha, + idempotency_key=args.idempotency_key, + repo_slug=args.slug, + ) + print(json.dumps(result.to_dict(), indent=2)) + return 0 if result.status == "applied" else 1 + + if args.command == "sync": + from repo_manager.projection_sync import sync_repository_projection + + result = sync_repository_projection( + Path(args.path), + api_base=args.api_base, + repo_slug=args.slug, + expected_instance_label=args.expected_instance_label, + commit_identifiers=not args.no_commit_identifiers, + push=args.push, + acknowledge_retirements=args.acknowledge_retirements, + queue_on_unavailable=not args.no_queue, + ) + print(json.dumps(result, indent=2)) + return 0 if result.get("ok") else 1 + if args.command == "workplan": from repo_manager.commands.workplan import mutate_workplan diff --git a/src/repo_manager/commands/task.py b/src/repo_manager/commands/task.py new file mode 100644 index 0000000..053cf16 --- /dev/null +++ b/src/repo_manager/commands/task.py @@ -0,0 +1,424 @@ +"""Governed file-backed task and daily ad-hoc mutations.""" + +from __future__ import annotations + +import json +import re +import uuid +from datetime import date +from pathlib import Path +from typing import Any + +from repo_manager import dual_run, idempotency +from repo_manager.commands.workplan import CommandResult, create_workplan +from repo_manager.gitops import GitError, commit_paths, head_sha, push_ff +from repo_manager.identifiers import derive_work_record_uuid, load_fleet_namespace +from repo_manager.index_store import append_event, default_index_path, save_index +from repo_manager.observe import observe_repository +from repo_manager.parse.workplan import _TASK_BLOCK_RE, _parse_yaml_block +from repo_manager.record_identity import classify_record_id +from repo_manager.time import utc_today + +VALID_TASK_STATUSES = frozenset({"wait", "todo", "progress", "done", "cancel"}) +VALID_TASK_PRIORITIES = frozenset({"low", "medium", "high", "critical"}) + + +def _result( + command: str, + status: str, + correlation_id: str, + *, + evidence: dict[str, Any] | None = None, + code: str | None = None, + message: str | None = None, +) -> CommandResult: + error = None + if code and message: + error = {"code": code, "message": message, "correlation_id": correlation_id} + return CommandResult( + command=command, + status=status, + correlation_id=correlation_id, + evidence=evidence or {"status": status}, + error=error, + ) + + +def _is_uuid(value: str) -> bool: + try: + uuid.UUID(value) + return True + except ValueError: + return False + + +def _find_record(index: Any, kind: str, identifier: str) -> Any | None: + by_uuid = _is_uuid(identifier) + for record in index.work_records: + if record.kind != kind: + continue + if not by_uuid and record.id == identifier: + return record + if by_uuid and record.uuid and record.uuid.lower() == identifier.lower(): + return record + return None + + +def _render_scalar(key: str, value: Any) -> str: + if isinstance(value, bool): + return "true" if value else "false" + if key in {"status", "priority"}: + return str(value) + return json.dumps(str(value), ensure_ascii=False) + + +def _patch_task_fields( + text: str, + *, + canonical_id: str, + updates: dict[str, Any], +) -> tuple[str, bool]: + matches = 0 + changed = False + + def replace(match: re.Match[str]) -> str: + nonlocal matches, changed + raw = match.group(1) + meta = _parse_yaml_block(raw.strip()) + if str(meta.get("id") or "") != canonical_id: + return match.group(0) + matches += 1 + patched = raw + for key, value in updates.items(): + rendered = _render_scalar(key, value) + pattern = re.compile(rf"^(?P{re.escape(key)}:\s*).*$", re.MULTILINE) + if pattern.search(patched): + candidate = pattern.sub(rf"\g{rendered}", patched, count=1) + else: + candidate = f"{patched.rstrip()}\n{key}: {rendered}\n" + if candidate != patched: + changed = True + patched = candidate + return f"```task\n{patched.strip()}\n```" + + result = _TASK_BLOCK_RE.sub(replace, text) + if matches != 1: + raise ValueError(f"{canonical_id}: expected exactly one task block, found {matches}") + return result, changed + + +def _next_task_id(parent_id: str, index: Any) -> str: + pattern = re.compile(rf"^{re.escape(parent_id)}-T(?P[0-9]{{2,3}})$") + numbers = [ + int(match.group("number")) + for record in index.work_records + if record.kind == "task" and record.id and (match := pattern.fullmatch(record.id)) + ] + number = max(numbers, default=0) + 1 + width = 2 if number < 100 else 3 + candidate = f"{parent_id}-T{number:0{width}d}" + if classify_record_id("task", candidate) != "canonical": + raise ValueError(f"no canonical task identifier remains under {parent_id}") + return candidate + + +def mutate_task( + repo_root: Path, + task_id: str | None, + *, + operation: str, + workplan_id: str | None = None, + title: str | None = None, + description: str | None = None, + status: str | None = None, + priority: str | None = None, + needs_human: bool | None = None, + intervention_note: str | None = None, + blocking_reason: str | None = None, + correlation_id: str | None = None, + reason: str = "rmgr command", + commit: bool = True, + push: bool = False, + expected_head_sha: str | None = None, + idempotency_key: str | None = None, + repo_slug: str | None = None, +) -> CommandResult: + command = f"repo.work.{operation}_task" + correlation_id = correlation_id or str(uuid.uuid4()) + repo_root = repo_root.resolve() + payload = { + "repo_root": str(repo_root), + "task_id": task_id, + "operation": operation, + "workplan_id": workplan_id, + "title": title, + "description": description, + "status": status, + "priority": priority, + "needs_human": needs_human, + "intervention_note": intervention_note, + "blocking_reason": blocking_reason, + "expected_head_sha": expected_head_sha, + } + payload_hash = idempotency.payload_hash(payload) + if idempotency_key and (prior := idempotency.get(idempotency_key)): + if prior.get("payload_hash") != payload_hash: + return _result( + command, "rejected", correlation_id, code="conflict", + message="idempotency key reused with different payload", + ) + previous = prior.get("result") or {} + return CommandResult( + command=previous.get("command", command), + status=previous.get("status", "applied"), + correlation_id=previous.get("correlation_id", correlation_id), + evidence=previous.get("evidence") or {"status": "applied", "replay": True}, + error=previous.get("error"), + ) + + if operation not in {"add", "update"}: + return _result(command, "rejected", correlation_id, code="validation_error", message="invalid task operation") + if status is not None and status not in VALID_TASK_STATUSES: + return _result(command, "rejected", correlation_id, code="validation_error", message=f"invalid task status {status!r}") + if priority is not None and priority not in VALID_TASK_PRIORITIES: + return _result(command, "rejected", correlation_id, code="validation_error", message=f"invalid task priority {priority!r}") + if needs_human is True and not intervention_note: + return _result(command, "rejected", correlation_id, code="validation_error", message="intervention_note is required when needs_human is true") + + current_head = head_sha(repo_root) + if expected_head_sha and current_head and expected_head_sha != current_head: + result = _result( + command, "rejected", correlation_id, + evidence={"status": "rejected", "head_sha": current_head}, + code="precondition_failed", + message=f"expected_head_sha {expected_head_sha} != {current_head}", + ) + if result.error: + result.error["retryable"] = True + return result + + snapshot, index = observe_repository(repo_root, slug=repo_slug) + slug = repo_slug or snapshot.get("slug") + if operation == "add": + if not workplan_id: + return _result(command, "rejected", correlation_id, code="validation_error", message="add requires workplan_id") + parent = _find_record(index, "workplan", workplan_id) + if parent is None or not parent.id: + return _result(command, "rejected", correlation_id, code="not_found", message=f"workplan {workplan_id!r} not found") + if not title or not title.strip(): + return _result(command, "rejected", correlation_id, code="validation_error", message="add requires a title") + canonical_id = task_id or _next_task_id(parent.id, index) + if classify_record_id("task", canonical_id) != "canonical": + return _result(command, "rejected", correlation_id, code="validation_error", message=f"invalid canonical task id {canonical_id!r}") + if _find_record(index, "task", canonical_id) is not None: + return _result(command, "rejected", correlation_id, code="conflict", message=f"task {canonical_id!r} exists") + if not canonical_id.startswith(parent.id + "-T"): + return _result(command, "rejected", correlation_id, code="validation_error", message="task id must be qualified by its workplan id") + path = repo_root / parent.source_path + task_uuid = derive_work_record_uuid(load_fleet_namespace(), canonical_id) + task_status = status or "todo" + task_priority = priority or "medium" + fields = [ + f"id: {canonical_id}", + f"status: {task_status}", + f"priority: {task_priority}", + f'state_hub_task_id: "{task_uuid}"', + ] + if needs_human is not None: + fields.append(f"needs_human: {'true' if needs_human else 'false'}") + if intervention_note: + fields.append(f"intervention_note: {json.dumps(intervention_note, ensure_ascii=False)}") + if blocking_reason: + fields.append(f"blocking_reason: {json.dumps(blocking_reason, ensure_ascii=False)}") + addition = ( + f"\n\n## {title.strip()}\n\n```task\n" + + "\n".join(fields) + + "\n```\n\n" + + ((description or "").strip() or "Task details to be completed.") + + "\n" + ) + path.write_text(path.read_text(encoding="utf-8").rstrip() + addition, encoding="utf-8") + changes = {"operation": "add", "task_id": canonical_id, "workplan_id": parent.id} + else: + if not task_id: + return _result(command, "rejected", correlation_id, code="validation_error", message="update requires task_id") + record = _find_record(index, "task", task_id) + if record is None or not record.id: + return _result(command, "rejected", correlation_id, code="not_found", message=f"task {task_id!r} not found") + updates = { + key: value + for key, value in { + "title": title.strip() if title else None, + "description": description.strip() if description else None, + "status": status, + "priority": priority, + "needs_human": needs_human, + "intervention_note": intervention_note, + "blocking_reason": blocking_reason, + }.items() + if value is not None + } + if not updates: + return _result(command, "rejected", correlation_id, code="validation_error", message="update requires at least one field") + existing_human = bool(record.extra.get("needs_human")) + resulting_human = updates.get("needs_human", existing_human) + resulting_note = updates.get("intervention_note", record.extra.get("intervention_note")) + if resulting_human and not resulting_note: + return _result(command, "rejected", correlation_id, code="validation_error", message="intervention_note is required when needs_human is true") + path = repo_root / record.source_path + patched, changed = _patch_task_fields( + path.read_text(encoding="utf-8"), canonical_id=record.id, updates=updates + ) + if not changed: + result = _result( + command, "applied", correlation_id, + evidence={"status": "applied", "git_sha": current_head, "files_touched": [], "noop": True}, + ) + if idempotency_key: + idempotency.put(idempotency_key, payload_hash, result.to_dict()) + return result + path.write_text(patched, encoding="utf-8") + canonical_id = record.id + changes = {"operation": "update", "task_id": canonical_id, "fields": sorted(updates)} + + relative_path = str(path.relative_to(repo_root)) + git_sha = None + push_ok = None + push_message = None + if commit: + try: + git_sha = commit_paths( + repo_root, + [relative_path], + message=( + f"{command} {canonical_id}\n\ncorrelation_id: {correlation_id}\n" + f"reason: {reason}\nsource: repo-manager\n" + ), + ) + except GitError as exc: + return _result(command, "failed", correlation_id, evidence={"status": "failed", "files_touched": [relative_path]}, code="internal", message=str(exc)) + if push: + push_ok, push_message = push_ff(repo_root) + + _snapshot_after, index_after = observe_repository(repo_root, slug=slug) + append_event( + index_after, + { + "type": "repo.command.applied", + "command": command, + "correlation_id": correlation_id, + "git_sha": git_sha, + **changes, + }, + ) + save_index(index_after) + dual_run.record_mutation( + source="repo-manager", + kind=f"task_{operation}", + repo_slug=slug, + detail={**changes, "git_sha": git_sha, "correlation_id": correlation_id}, + ) + evidence = { + "status": "applied", + "task_id": canonical_id, + "git_sha": git_sha, + "files_touched": [relative_path], + "index_path": str(default_index_path(repo_root)), + "source": "repo-manager", + } + if push: + evidence.update({"push_ok": push_ok, "push_message": push_message}) + result = _result(command, "applied", correlation_id, evidence=evidence) + if idempotency_key: + idempotency.put(idempotency_key, payload_hash, result.to_dict()) + return result + + +def add_adhoc_task( + repo_root: Path, + title: str, + description: str, + *, + prefix: str | None = None, + on_date: date | None = None, + status: str = "todo", + priority: str = "medium", + owner: str = "codex", + correlation_id: str | None = None, + reason: str = "rmgr adhoc add", + commit: bool = True, + push: bool = False, + expected_head_sha: str | None = None, + idempotency_key: str | None = None, + repo_slug: str | None = None, +) -> CommandResult: + """Create/reuse today's qualified ad-hoc workplan and append one task.""" + correlation_id = correlation_id or str(uuid.uuid4()) + repo_root = repo_root.resolve() + payload = { + "repo_root": str(repo_root), "title": title, "description": description, + "prefix": prefix, "on_date": str(on_date) if on_date else None, + "status": status, "priority": priority, "expected_head_sha": expected_head_sha, + } + payload_hash = idempotency.payload_hash(payload) + if idempotency_key and (prior := idempotency.get(idempotency_key)): + if prior.get("payload_hash") != payload_hash: + return _result("repo.work.add_adhoc_task", "rejected", correlation_id, code="conflict", message="idempotency key reused with different payload") + previous = prior.get("result") or {} + return CommandResult( + command=previous.get("command", "repo.work.add_adhoc_task"), + status=previous.get("status", "applied"), + correlation_id=previous.get("correlation_id", correlation_id), + evidence=previous.get("evidence") or {"status": "applied", "replay": True}, + error=previous.get("error"), + ) + + snapshot, index = observe_repository(repo_root, slug=repo_slug) + prefixes = { + record.id.split("-WP-", 1)[0] + for record in index.work_records + if record.kind == "workplan" and record.id and "-WP-" in record.id + } + selected = (prefix or "").strip().upper().removesuffix("-WP") + if not selected: + if len(prefixes) != 1: + return _result("repo.work.add_adhoc_task", "rejected", correlation_id, code="validation_error", message=f"could not infer one workplan prefix; candidates={sorted(prefixes)!r}; pass --prefix") + selected = next(iter(prefixes)) + day = on_date or utc_today() + workplan_id = f"{selected}-WP-ADHOC-{day.isoformat()}" + parent = _find_record(index, "workplan", workplan_id) + if parent is None: + created = create_workplan( + repo_root, + workplan_id, + f"Ad Hoc — {day.isoformat()}", + f"Small opportunistic fixes completed on {day.isoformat()}.", + owner=owner, + status="active", + commit=False, + expected_head_sha=expected_head_sha, + repo_slug=repo_slug or snapshot.get("slug"), + ) + if created.status != "applied": + return created + result = mutate_task( + repo_root, + None, + operation="add", + workplan_id=workplan_id, + title=title, + description=description, + status=status, + priority=priority, + correlation_id=correlation_id, + reason=reason, + commit=commit, + push=push, + expected_head_sha=expected_head_sha, + repo_slug=repo_slug or snapshot.get("slug"), + ) + result.command = "repo.work.add_adhoc_task" + result.evidence["workplan_id"] = workplan_id + if idempotency_key and result.status == "applied": + idempotency.put(idempotency_key, payload_hash, result.to_dict()) + return result diff --git a/src/repo_manager/commands/workplan.py b/src/repo_manager/commands/workplan.py index 61450f5..0211022 100644 --- a/src/repo_manager/commands/workplan.py +++ b/src/repo_manager/commands/workplan.py @@ -12,14 +12,15 @@ from typing import Any from repo_manager import dual_run, idempotency from repo_manager.gitops import GitError, commit_paths, head_sha, push_ff +from repo_manager.identifiers import derive_work_record_uuid, load_fleet_namespace from repo_manager.index_store import append_event, default_index_path, save_index from repo_manager.observe import observe_repository +from repo_manager.record_identity import classify_record_id from repo_manager.time import utc_today VALID_WORKPLAN_STATUSES = frozenset( {"proposed", "ready", "active", "blocked", "backlog", "finished", "archived"} ) -_WORKPLAN_ID_RE = re.compile(r"^[A-Z][A-Z0-9]*-WP-[0-9]{4,}$") _FRONTMATTER_RE = re.compile(r"\A---\r?\n(?P.*?)\r?\n---(?P\r?\n.*)?\Z", re.DOTALL) @@ -123,6 +124,7 @@ def _create_text( topic_slug: str, ) -> str: today = _today().isoformat() + workplan_uuid = derive_work_record_uuid(load_fleet_namespace(), workplan_id) return ( "---\n" f"id: {workplan_id}\n" @@ -135,6 +137,7 @@ def _create_text( f"topic_slug: {topic_slug}\n" f"created: {_quoted(today)}\n" f"updated: {_quoted(today)}\n" + f"state_hub_workstream_id: {_quoted(str(workplan_uuid))}\n" "---\n\n" f"# {title}\n\n" "## Goal\n\n" @@ -241,7 +244,7 @@ def mutate_workplan( changes: dict[str, Any] if operation == "create": - if not _WORKPLAN_ID_RE.fullmatch(workplan_id): + if classify_record_id("workplan", workplan_id) != "canonical": return _rejected( command, correlation_id, @@ -265,8 +268,23 @@ def mutate_workplan( "validation_error", f"invalid workplan status {create_status!r}", ) - name = filename or f"{workplan_id}-{_slugify(title)}.md" - if Path(name).name != name or not name.endswith(".md") or not name.startswith(f"{workplan_id}-"): + is_adhoc = "-WP-ADHOC-" in workplan_id + adhoc_date = workplan_id.rsplit("-WP-ADHOC-", 1)[-1] + name = filename or ( + f"ADHOC-{adhoc_date}.md" + if is_adhoc + else f"{workplan_id}-{_slugify(title)}.md" + ) + expected_adhoc_name = f"ADHOC-{adhoc_date}.md" + safe_name = ( + Path(name).name == name + and name.endswith(".md") + and ( + name.startswith(f"{workplan_id}-") + or (is_adhoc and name == expected_adhoc_name) + ) + ) + if not safe_name: return _rejected(command, correlation_id, "validation_error", "unsafe workplan filename") path = repo_root / "workplans" / name if path.exists(): diff --git a/src/repo_manager/identifiers.py b/src/repo_manager/identifiers.py index 1beb06e..0d38d8d 100644 --- a/src/repo_manager/identifiers.py +++ b/src/repo_manager/identifiers.py @@ -25,6 +25,7 @@ from repo_manager.parse.workplan import ( parse_workplan_file, ) from repo_manager.prefix_registry import iter_repo_roots +from repo_manager.record_identity import classify_record_id, scan_record_identities from repo_manager.time import utc_now_text DERIVATION_NAMESPACE_UUID = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009") @@ -33,7 +34,6 @@ FLEET_NAMESPACE_SCHEMA = "repo-manager.fleet-namespace.v1" DEFAULT_NAMESPACE_CONTRACT = Path(__file__).resolve().parents[2] / "config" / "fleet-namespace.yaml" LIVE_WORKPLAN_STATUSES = frozenset({"proposed", "ready", "active", "blocked", "backlog"}) _NAMESPACE_RE = re.compile(r"^[a-z0-9][a-z0-9.-]{0,62}$") -_RECORD_ID_RE = re.compile(r"^[A-Z][A-Z0-9-]*-WP-[0-9]{4}(?:-T[0-9]{2,})?$") def load_fleet_namespace(path: Path = DEFAULT_NAMESPACE_CONTRACT) -> str: @@ -52,7 +52,15 @@ def derivation_name(namespace: str, identifier: str) -> str: identifier = identifier.strip() if not _NAMESPACE_RE.fullmatch(namespace): raise ValueError("namespace must be lowercase DNS-label style") - if not _RECORD_ID_RE.fullmatch(identifier): + canonical_kind = next( + ( + kind + for kind in ("workplan", "task") + if classify_record_id(kind, identifier) == "canonical" + ), + None, + ) + if canonical_kind is None: raise ValueError("identifier must be a canonical workplan or task id") return f"{namespace}\n{identifier}" @@ -732,6 +740,101 @@ def _atomic_write(path: Path, content: str) -> None: os.unlink(temporary_name) +def ensure_missing_work_record_identifiers( + repo_root: Path, + *, + namespace: str | None = None, + execute: bool = True, +) -> dict[str, Any]: + """Assign only missing canonical UUIDs, deterministically, in one repository. + + Existing UUIDs are deliberately preserved. Replacing a legacy UUID is a + separately sealed migration; ordinary registration must never perform it. + Every source file is validated and rendered before any file is changed. + """ + repo_root = repo_root.resolve() + namespace = namespace or load_fleet_namespace() + identity = scan_record_identities(repo_root) + if identity["invalid_identifiers"] or identity["identity_collisions"]: + raise ValueError( + "work-record identities are not safe to derive: " + f"invalid={identity['invalid_identifiers']!r}, " + f"collisions={identity['identity_collisions']!r}" + ) + + rendered: dict[Path, str] = {} + assignments: list[dict[str, str]] = [] + skipped: list[dict[str, str]] = [] + for path in iter_workplan_files(repo_root): + parsed = parse_workplan_file(path, repo_root=repo_root) + text = path.read_text(encoding="utf-8") + changed = False + if parsed.id and not parsed.state_hub_workstream_id: + if classify_record_id("workplan", parsed.id) != "canonical": + skipped.append( + {"kind": "workplan", "record_id": parsed.id, "path": parsed.path} + ) + else: + derived = str(derive_work_record_uuid(namespace, parsed.id)) + text = _patch_workplan_identifier( + text, + record_id=parsed.id, + expected=None, + target=derived, + ) + assignments.append( + { + "kind": "workplan", + "record_id": parsed.id, + "uuid": derived, + "path": parsed.path, + } + ) + changed = True + for task in parsed.tasks: + if not task.id or task.state_hub_task_id: + continue + if classify_record_id("task", task.id) != "canonical": + skipped.append({"kind": "task", "record_id": task.id, "path": parsed.path}) + continue + derived = str(derive_work_record_uuid(namespace, task.id)) + text = _patch_task_identifier( + text, + record_id=task.id, + expected=None, + target=derived, + ) + assignments.append( + { + "kind": "task", + "record_id": task.id, + "uuid": derived, + "path": parsed.path, + } + ) + changed = True + if changed: + rendered[path] = text + + if skipped: + raise ValueError( + "missing UUIDs on grandfathered or noncanonical records require " + f"an explicit identity decision: {skipped!r}" + ) + if execute: + for path, content in rendered.items(): + _atomic_write(path, content) + return { + "schema": "repo-manager.identifier-assignment.v1", + "ok": True, + "executed": execute, + "repo": repo_root.name, + "namespace": namespace, + "assignments": assignments, + "files_changed": sorted(str(path.relative_to(repo_root)) for path in rendered), + } + + def migrate_repository_identifier_files( plan: dict[str, Any], *, diff --git a/src/repo_manager/projection_sync.py b/src/repo_manager/projection_sync.py new file mode 100644 index 0000000..23fc2d4 --- /dev/null +++ b/src/repo_manager/projection_sync.py @@ -0,0 +1,239 @@ +"""Fast, guarded trigger for State Hub's forge-derived repository projection.""" + +from __future__ import annotations + +import json +import subprocess +import tempfile +import time +from pathlib import Path +from typing import Any + +import httpx + +from repo_manager.gitops import GitError, commit_paths, head_sha, push_ff +from repo_manager.identifiers import ensure_missing_work_record_identifiers + + +def _git(repo: Path, *args: str) -> subprocess.CompletedProcess[str]: + return subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True, check=False) + + +def _ahead_behind(repo: Path) -> dict[str, Any]: + upstream = _git(repo, "rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}") + if upstream.returncode != 0: + raise ValueError("current branch has no upstream; central can only derive pushed work") + counts = _git(repo, "rev-list", "--left-right", "--count", "@{u}...HEAD") + if counts.returncode != 0: + raise ValueError(counts.stderr.strip() or "could not compare HEAD with upstream") + try: + behind, ahead = (int(value) for value in counts.stdout.split()) + except ValueError as exc: + raise ValueError("could not parse Git ahead/behind state") from exc + return {"upstream": upstream.stdout.strip(), "behind": behind, "ahead": ahead} + + +def _dirty_authoritative_paths(repo: Path) -> list[str]: + result = _git(repo, "status", "--porcelain", "--", "workplans") + if result.returncode != 0: + raise ValueError(result.stderr.strip() or "could not inspect workplan files") + return [line[3:] for line in result.stdout.splitlines() if len(line) > 3] + + +def _pending_path(repo: Path) -> Path: + resolved = _git( + repo, "rev-parse", "--git-path", "repo-manager/pending-work-record-projection.json" + ) + if resolved.returncode != 0 or not resolved.stdout.strip(): + raise ValueError("could not resolve the repository-local Git state directory") + path = Path(resolved.stdout.strip()) + return path if path.is_absolute() else repo / path + + +def _write_pending(repo: Path, payload: dict[str, Any]) -> Path: + path = _pending_path(repo) + path.parent.mkdir(parents=True, exist_ok=True) + temporary: Path | None = None + try: + with tempfile.NamedTemporaryFile( + mode="w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False + ) as handle: + temporary = Path(handle.name) + handle.write(json.dumps(payload, indent=2) + "\n") + temporary.replace(path) + temporary = None + finally: + if temporary is not None: + temporary.unlink(missing_ok=True) + return path + + +def sync_repository_projection( + repo_root: Path, + *, + api_base: str, + repo_slug: str | None = None, + expected_instance_label: str | None = "railliance01", + commit_identifiers: bool = True, + push: bool = False, + acknowledge_retirements: bool = False, + queue_on_unavailable: bool = True, + transport: httpx.BaseTransport | None = None, +) -> dict[str, Any]: + """Assign missing IDs, prove the commit is pushed, then reconcile once.""" + started = time.perf_counter() + repo_root = repo_root.resolve() + slug = repo_slug or repo_root.name + + dirty = _dirty_authoritative_paths(repo_root) + if dirty: + return { + "ok": False, + "status": "pending_commit", + "error": "authoritative workplan files have uncommitted changes", + "dirty_paths": dirty, + } + + assignments = ensure_missing_work_record_identifiers(repo_root) + if assignments["files_changed"] and commit_identifiers: + try: + commit_paths( + repo_root, + assignments["files_changed"], + message=( + "repo.work.assign_missing_identifiers\n\n" + "source: repo-manager\nreason: deterministic projection registration\n" + ), + ) + except GitError as exc: + return {"ok": False, "status": "failed", "error": str(exc), "identifiers": assignments} + + dirty = _dirty_authoritative_paths(repo_root) + if dirty: + return { + "ok": False, + "status": "pending_commit", + "error": "authoritative workplan files have uncommitted changes", + "dirty_paths": dirty, + "identifiers": assignments, + } + + try: + git = _ahead_behind(repo_root) + except ValueError as exc: + return {"ok": False, "status": "failed", "error": str(exc), "identifiers": assignments} + if git["behind"]: + return { + "ok": False, + "status": "behind_upstream", + "error": "branch is behind upstream; reconcile Git before projection", + "git": git, + } + if git["ahead"] and push: + pushed, message = push_ff(repo_root) + if not pushed: + return { + "ok": False, + "status": "push_failed", + "error": message, + "git": git, + } + git = _ahead_behind(repo_root) + if git["ahead"]: + return { + "ok": True, + "status": "pending_push", + "message": "commit is not yet visible to the authoritative forge", + "git": git, + "identifiers": assignments, + } + + commit = head_sha(repo_root) + if not commit: + return {"ok": False, "status": "failed", "error": "repository HEAD is unavailable"} + + pending = { + "schema": "repo-manager.pending-work-record-projection.v1", + "repo_slug": slug, + "expected_commit": commit, + "api_base": api_base.rstrip("/"), + } + headers = { + "Idempotency-Key": f"rmgr-projection:{slug}:{commit}", + "X-StateHub-Source-Agent": "repo-manager", + } + try: + with httpx.Client( + base_url=api_base.rstrip("/"), + timeout=httpx.Timeout(120.0, connect=5.0), + follow_redirects=True, + transport=transport, + ) as client: + health = client.get("/state/health") + health.raise_for_status() + identity = health.json() + identity_ok = identity.get("instance_role") == "primary" and ( + expected_instance_label is None + or identity.get("instance_label") == expected_instance_label + ) + if not identity_ok: + if not queue_on_unavailable: + return { + "ok": False, + "status": "wrong_instance", + "instance": identity, + } + path = _write_pending(repo_root, {**pending, "reason": "wrong_instance"}) + return { + "ok": True, + "status": "queued", + "reason": "wrong_instance", + "instance": identity, + "pending_path": str(path), + } + response = client.post( + f"/repos/{slug}/work-record-projection/reconcile", + json={ + "expected_commit": commit, + "acknowledge_retirements": acknowledge_retirements, + }, + headers=headers, + ) + response.raise_for_status() + receipt = response.json() + except (httpx.ConnectError, httpx.TimeoutException, httpx.NetworkError) as exc: + if not queue_on_unavailable: + return {"ok": False, "status": "unavailable", "error": str(exc)} + path = _write_pending(repo_root, {**pending, "reason": "hub_unavailable"}) + return { + "ok": True, + "status": "queued", + "reason": "hub_unavailable", + "pending_path": str(path), + } + except httpx.HTTPStatusError as exc: + detail: Any + try: + detail = exc.response.json() + except ValueError: + detail = exc.response.text[:500] + return { + "ok": False, + "status": "rejected", + "http_status": exc.response.status_code, + "error": detail, + } + + _pending_path(repo_root).unlink(missing_ok=True) + outcome_status = receipt.get("outcome", {}).get("status", "applied") + return { + "ok": outcome_status in {"applied", "noop"}, + "status": outcome_status, + "repo_slug": slug, + "commit": commit, + "requests": 2, + "elapsed_ms": round((time.perf_counter() - started) * 1000, 1), + "git": git, + "identifiers": assignments, + "receipt": receipt, + } diff --git a/tests/test_fast_work_records.py b/tests/test_fast_work_records.py new file mode 100644 index 0000000..6dfd543 --- /dev/null +++ b/tests/test_fast_work_records.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import subprocess +from datetime import date +from pathlib import Path + +import httpx + +from repo_manager.commands.task import add_adhoc_task, mutate_task +from repo_manager.commands.workplan import create_workplan +from repo_manager.identifiers import ( + derive_work_record_uuid, + ensure_missing_work_record_identifiers, +) +from repo_manager.projection_sync import sync_repository_projection + + +def _git(repo: Path, *args: str) -> None: + subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True) + + +def _fixture(tmp_path: Path, *, remote: bool = False) -> Path: + repo = tmp_path / "pilot" + repo.mkdir() + _git(repo, "init") + _git(repo, "config", "user.email", "test@example.com") + _git(repo, "config", "user.name", "Test") + (repo / "workplans").mkdir() + (repo / ".repo-classification.yaml").write_text( + "repo_classification:\n category: tooling\n domain: infotech\n", + encoding="utf-8", + ) + (repo / "workplans" / "P-WP-0001-first.md").write_text( + """--- +id: P-WP-0001 +type: workplan +title: First +domain: infotech +repo: pilot +status: active +owner: codex +--- + +# First + +## Existing task + +```task +id: P-WP-0001-T01 +status: todo +priority: high +``` + +Existing task details. +""", + encoding="utf-8", + ) + _git(repo, "add", ".") + _git(repo, "commit", "-m", "seed") + if remote: + bare = tmp_path / "remote.git" + subprocess.run(["git", "init", "--bare", str(bare)], check=True, capture_output=True) + _git(repo, "remote", "add", "origin", str(bare)) + _git(repo, "push", "-u", "origin", "HEAD") + return repo + + +def test_adhoc_ids_are_canonical_and_stable() -> None: + identifier = "P-WP-ADHOC-2026-08-30" + task_id = f"{identifier}-T01" + assert derive_work_record_uuid("helixforge", identifier) == derive_work_record_uuid( + "helixforge", identifier + ) + assert derive_work_record_uuid("helixforge", task_id).version == 5 + + +def test_missing_assignment_preserves_existing_uuid(tmp_path: Path) -> None: + repo = _fixture(tmp_path) + existing = "11111111-1111-4111-8111-111111111111" + path = repo / "workplans" / "P-WP-0001-first.md" + text = path.read_text(encoding="utf-8").replace( + "owner: codex\n", f'owner: codex\nstate_hub_workstream_id: "{existing}"\n' + ) + path.write_text(text, encoding="utf-8") + + result = ensure_missing_work_record_identifiers(repo) + + assert result["ok"] is True + assert {item["record_id"] for item in result["assignments"]} == {"P-WP-0001-T01"} + assert existing in path.read_text(encoding="utf-8") + + +def test_create_adhoc_uses_conventional_filename_and_uuid(tmp_path: Path) -> None: + repo = _fixture(tmp_path) + result = create_workplan( + repo, + "P-WP-ADHOC-2026-08-30", + "Ad Hoc — 2026-08-30", + "Small fixes.", + ) + + assert result.status == "applied" + path = repo / "workplans" / "ADHOC-2026-08-30.md" + assert path.is_file() + assert str(derive_work_record_uuid("helixforge", "P-WP-ADHOC-2026-08-30")) in path.read_text( + encoding="utf-8" + ) + + +def test_task_add_update_and_adhoc_replay(tmp_path: Path, monkeypatch) -> None: + repo = _fixture(tmp_path) + monkeypatch.setenv("RM_IDEMPOTENCY_PATH", str(tmp_path / "idempotency.json")) + added = mutate_task( + repo, + None, + operation="add", + workplan_id="P-WP-0001", + title="Second task", + description="Do the second thing.", + priority="high", + ) + assert added.status == "applied" + assert added.evidence["task_id"] == "P-WP-0001-T02" + + updated = mutate_task( + repo, + "P-WP-0001-T02", + operation="update", + status="progress", + needs_human=True, + intervention_note="Review the external effect.", + ) + assert updated.status == "applied" + text = (repo / "workplans" / "P-WP-0001-first.md").read_text(encoding="utf-8") + assert "status: progress" in text + assert "needs_human: true" in text + + first = add_adhoc_task( + repo, + "Tiny fix", + "Complete the tiny fix.", + on_date=date(2026, 8, 30), + idempotency_key="adhoc-1", + ) + replay = add_adhoc_task( + repo, + "Tiny fix", + "Complete the tiny fix.", + on_date=date(2026, 8, 30), + idempotency_key="adhoc-1", + ) + assert first.status == replay.status == "applied" + assert first.evidence["task_id"] == replay.evidence["task_id"] + + +def test_sync_uses_two_requests_and_exact_pushed_commit(tmp_path: Path) -> None: + repo = _fixture(tmp_path, remote=True) + seen: list[str] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append(request.url.path) + if request.url.path == "/state/health": + return httpx.Response( + 200, + json={"status": "ok", "instance_role": "primary", "instance_label": "railliance01"}, + ) + payload = __import__("json").loads(request.content) + assert ( + payload["expected_commit"] + == subprocess.run( + ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True + ).stdout.strip() + ) + return httpx.Response( + 200, + json={ + "outcome": {"status": "applied", "counts": {"created": 1}}, + "derived_commit": payload["expected_commit"], + }, + ) + + result = sync_repository_projection( + repo, + api_base="http://hub.test", + push=True, + transport=httpx.MockTransport(handler), + ) + + assert result["ok"] is True + assert result["requests"] == 2 + assert seen == ["/state/health", "/repos/pilot/work-record-projection/reconcile"] + + +def test_sync_never_assigns_or_commits_over_dirty_workplans(tmp_path: Path) -> None: + repo = _fixture(tmp_path, remote=True) + path = repo / "workplans" / "P-WP-0001-first.md" + original = path.read_text(encoding="utf-8") + path.write_text(original.replace("First\n", "Locally edited\n", 1), encoding="utf-8") + + result = sync_repository_projection(repo, api_base="http://hub.test", push=True) + + assert result["ok"] is False + assert result["status"] == "pending_commit" + assert "state_hub_workstream_id" not in path.read_text(encoding="utf-8") + assert ( + subprocess.run( + ["git", "log", "-1", "--pretty=%s"], + cwd=repo, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + == "seed" + ) + + +def test_sync_queues_instead_of_writing_to_wrong_instance(tmp_path: Path) -> None: + repo = _fixture(tmp_path, remote=True) + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json={"status": "ok", "instance_role": "cache", "instance_label": "workstation"}, + ) + + result = sync_repository_projection( + repo, + api_base="http://cache.test", + push=True, + transport=httpx.MockTransport(handler), + ) + + assert result["ok"] is True + assert result["status"] == "queued" + assert result["reason"] == "wrong_instance" + assert Path(result["pending_path"]).is_file() diff --git a/workplans/RMGR-WP-0012-fast-work-record-mutation-and-projection.md b/workplans/RMGR-WP-0012-fast-work-record-mutation-and-projection.md new file mode 100644 index 0000000..1109476 --- /dev/null +++ b/workplans/RMGR-WP-0012-fast-work-record-mutation-and-projection.md @@ -0,0 +1,86 @@ +--- +id: RMGR-WP-0012 +type: workplan +title: "Fast work-record mutation and forge projection trigger" +domain: infotech +repo: repo-manager +status: active +owner: codex +topic_slug: infotech +created: "2026-08-30" +updated: "2026-08-30" +related: + - RMGR-WP-0005 + - RMGR-WP-0008 + - STATE-WP-0083 + - STATE-WP-0086 +state_hub_workstream_id: "c9690d0f-ed75-56a5-9738-39fe49856e2e" +--- + +# Fast work-record mutation and forge projection trigger + +## Goal + +Make ad-hoc, workplan, and task creation and updates deterministic and quick, +then ask the authoritative State Hub to derive one repository projection from +the pushed Forgejo commit. Repository files and the forge remain authoritative; +Repo Manager does not upload a workstation-built replacement projection. + +## Canonical identifiers and deterministic registration + +```task +id: RMGR-WP-0012-T01 +status: done +priority: high +state_hub_task_id: "7d8ecdbd-4350-54e7-8f6c-4692e1e0f3ea" +``` + +Use the work-record kind registry as the only identifier grammar. Accept +repository-qualified ad-hocs and assign missing deterministic UUIDv5 values +locally without replacing legacy UUIDs or requiring a registrar host. + +## Governed ad-hoc and task mutations + +```task +id: RMGR-WP-0012-T02 +status: done +priority: high +state_hub_task_id: "f6add501-4153-5df5-aacc-ebd444dca00c" +``` + +Add concurrency-safe, idempotent commands for daily ad-hoc task creation and +general task creation/update while retaining optimistic Git HEAD checks and +explicit push sealing. + +## Fast authoritative projection trigger + +```task +id: RMGR-WP-0012-T03 +status: done +priority: high +state_hub_task_id: "02d155c6-0baa-573a-9055-4aee91dcb246" +``` + +Add one sync command that validates local records, writes only missing derived +identifiers, verifies the branch is visible on its upstream, verifies the hub +is the railiance01 primary, and requests reconciliation against that exact +forge commit. + +## Regression, performance, and operator cutover + +```task +id: RMGR-WP-0012-T04 +status: progress +priority: high +state_hub_task_id: "723c77ab-c2a3-545f-94d2-174b6f983a67" +``` + +Cover canonical/ad-hoc grammar, retries, Git preconditions, incorrect hub +identity, exact-commit reconciliation, and concise receipts. Document the fast +path and retain registrar reconciliation only as a legacy migration tool. + +Implementation and local verification are complete: the Repo Manager suite is +green, the focused fast-path tests enforce two HTTP requests, and current agent +and repository-standard documentation points at `rmgr sync`. This task remains +in progress until the matching State Hub endpoint is deployed and one live +Forgejo-to-primary receipt establishes the production latency baseline.