tegwick
8b5634ff2a
Implement observe/reconcile/update-task-status CLI path: workplan parse, JSON projection index, git-backed task status writeback with correlation events, and E2E pytest plus evidence artifacts. Finish foundation workplan.
107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
"""Build repository observation + work-record index from files."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
|
|
from repo_manager.gitops import head_sha, is_git_repo
|
|
from repo_manager.index_store import RepoIndex, WorkRecordEntry, _now
|
|
from repo_manager.parse.workplan import iter_workplan_files, parse_workplan_file
|
|
|
|
|
|
def _slug_from_path(repo_root: Path) -> str:
|
|
return re.sub(r"[^a-z0-9]+", "-", repo_root.name.lower()).strip("-") or "repo"
|
|
|
|
|
|
def load_classification(repo_root: Path) -> dict | None:
|
|
path = repo_root / ".repo-classification.yaml"
|
|
if not path.is_file():
|
|
return None
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
if isinstance(data, dict) and "repo_classification" in data:
|
|
return data["repo_classification"]
|
|
return data if isinstance(data, dict) else None
|
|
|
|
|
|
def observe_repository(repo_root: Path, *, slug: str | None = None) -> tuple[dict, RepoIndex]:
|
|
"""Return (RepositorySnapshot-like dict, rebuilt RepoIndex)."""
|
|
repo_root = repo_root.resolve()
|
|
slug = slug or _slug_from_path(repo_root)
|
|
classification = load_classification(repo_root)
|
|
intent_path = None
|
|
description = None
|
|
for name in ("INTENT.md", "GOAL.md"):
|
|
p = repo_root / name
|
|
if p.is_file():
|
|
intent_path = name
|
|
# first non-empty line of body as weak description
|
|
text = p.read_text(encoding="utf-8")
|
|
for line in text.splitlines():
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and not line.startswith("---") and not line.startswith(">"):
|
|
description = line[:200]
|
|
break
|
|
break
|
|
|
|
records: list[WorkRecordEntry] = []
|
|
for path in iter_workplan_files(repo_root):
|
|
wp = parse_workplan_file(path, repo_root=repo_root)
|
|
records.append(
|
|
WorkRecordEntry(
|
|
kind="workplan",
|
|
id=wp.id,
|
|
status=wp.status,
|
|
title=wp.title,
|
|
source_path=wp.path,
|
|
uuid=wp.state_hub_workstream_id,
|
|
)
|
|
)
|
|
for task in wp.tasks:
|
|
records.append(
|
|
WorkRecordEntry(
|
|
kind="task",
|
|
id=task.id,
|
|
status=task.status,
|
|
title=task.title,
|
|
source_path=wp.path,
|
|
uuid=task.state_hub_task_id,
|
|
parent_id=wp.id,
|
|
)
|
|
)
|
|
|
|
sha = head_sha(repo_root) if is_git_repo(repo_root) else None
|
|
index = RepoIndex(
|
|
slug=slug,
|
|
repo_root=str(repo_root),
|
|
head_sha=sha,
|
|
observed_at=_now(),
|
|
work_records=records,
|
|
)
|
|
|
|
snapshot = {
|
|
"api_version": "0.1",
|
|
"slug": slug,
|
|
"lifecycle": "active",
|
|
"domain": (classification or {}).get("domain"),
|
|
"classification": classification,
|
|
"purpose": {
|
|
"description": description,
|
|
"intent_path": intent_path,
|
|
},
|
|
"locations": {
|
|
"repo_root": str(repo_root),
|
|
},
|
|
"revision": {
|
|
"head_sha": sha,
|
|
"observed_at": index.observed_at,
|
|
},
|
|
"index": {
|
|
"workplan_count": sum(1 for r in records if r.kind == "workplan"),
|
|
"task_count": sum(1 for r in records if r.kind == "task"),
|
|
"record_count": len(records),
|
|
},
|
|
}
|
|
return snapshot, index
|