tegwick
58414404d6
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a053ff-1d6f-7fe2-ac1c-a6eb40a42a0c
424 lines
17 KiB
Python
424 lines
17 KiB
Python
"""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<prefix>{re.escape(key)}:\s*).*$", re.MULTILINE)
|
|
if pattern.search(patched):
|
|
candidate = pattern.sub(rf"\g<prefix>{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<number>[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
|