coulomb/the-custodian
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Improved workplan dependency management facilities

Browse Source
This commit is contained in:
Bernd Worsch
2026-05-04 11:45:24 +02:00
parent 4d6164e81b
commit bfed370a6e
10 changed files with 380 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

163
state-hub/scripts/consistency_check.py
View File

@@ -22,6 +22,8 @@ Checks:
C-15 task-db-ahead WARN Yes DB task status is ahead of file — regression prevented; writeback syncs file
C-16 repo-behind-remote WARN No Local repo is behind remote tracking branch — --fix skipped to avoid clobbering remote progress
C-17 repo-ahead-push-failed WARN No Local repo has unpushed commits and push failed — writes skipped to prevent runaway divergence
C-19 workstream-planning-drift WARN Yes planning_priority/planning_order differs between file and DB
C-20 workstream-dependency-missing WARN Yes Workplan dependency frontmatter missing from DB graph
Usage:
python scripts/consistency_check.py --repo SLUG [--fix] [--no-writeback] [--json] [--api-base URL]
@@ -69,6 +71,7 @@ _ARCHIVED_WP_RE = re.compile(r"^\d{6}-(.+\.md)$")
VALID_WP_STATUSES = {"active", "completed", "archived"}
VALID_TASK_STATUSES = {"todo", "in_progress", "blocked", "done", "cancelled"}
VALID_TASK_PRIORITIES = {"low", "medium", "high", "critical"}
VALID_DEP_RELATIONSHIPS = {"blocks", "starts_after", "informs", "soft_dependency"}
# Workplan files use task-style vocabulary ("done"); the DB workstream API uses
# "completed". This map translates file values to DB values before comparison
@@ -214,6 +217,25 @@ def get_tasks_from_workplan(meta: dict, body: str) -> list[dict]:
return []
def _as_list(value: Any) -> list[str]:
if value is None:
return []
if isinstance(value, list):
return [str(item).strip().strip('"') for item in value if str(item).strip()]
if isinstance(value, str):
return [item.strip().strip('"') for item in value.split(",") if item.strip()]
return [str(value).strip().strip('"')]
def _as_int_or_none(value: Any) -> int | None:
if value in (None, "", "~", "null", "None", "none"):
return None
try:
return int(value)
except (TypeError, ValueError):
return None
# ---------------------------------------------------------------------------
# File update helpers
# ---------------------------------------------------------------------------
@@ -511,6 +533,22 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
if wp_file.parent == workplans_dir:
active_file_ws_ids.add(ws_id)
workplan_id_to_ws_id: dict[str, str] = {}
task_file_id_to_sh_id: dict[str, str] = {}
for _wp_file, meta, body in workplan_infos:
mapped_ws_id = str(meta.get("state_hub_workstream_id", "")).strip().strip('"')
wp_id = str(meta.get("id", "")).strip()
if wp_id and mapped_ws_id:
workplan_id_to_ws_id[wp_id] = mapped_ws_id
for task in get_tasks_from_workplan(meta, body):
if task.get("_parse_error"):
continue
task_file_id = str(task.get("id", "")).strip()
raw_sh = task.get("state_hub_task_id")
task_sh_id = "" if raw_sh is None else str(raw_sh).strip().strip('"')
if task_file_id and task_sh_id and task_sh_id not in ("~", "null", "None", "none"):
task_file_id_to_sh_id[task_file_id] = task_sh_id
# Per-workplan checks
for wp_file, meta, body in workplan_infos:
fname = workplan_display_path(repo_dir, wp_file)
@@ -618,6 +656,38 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
_fix_context={"ws_id": ws_id, "field": "title", "value": file_title},
)
planning_priority = str(meta.get("planning_priority", "")).strip() or None
if planning_priority != (ws.get("planning_priority") or None):
report.add(
severity="WARN", check_id="C-19",
message=(
f"Planning priority drift in '{ws.get('slug')}': "
f"file={planning_priority!r} db={ws.get('planning_priority')!r} (file wins)"
),
file_path=fname,
db_id=ws_id,
file_value=planning_priority,
db_value=ws.get("planning_priority"),
fixable=True,
_fix_context={"ws_id": ws_id, "field": "planning_priority", "value": planning_priority},
)
planning_order = _as_int_or_none(meta.get("planning_order"))
if planning_order != ws.get("planning_order"):
report.add(
severity="WARN", check_id="C-19",
message=(
f"Planning order drift in '{ws.get('slug')}': "
f"file={planning_order!r} db={ws.get('planning_order')!r} (file wins)"
),
file_path=fname,
db_id=ws_id,
file_value=planning_order,
db_value=ws.get("planning_order"),
fixable=True,
_fix_context={"ws_id": ws_id, "field": "planning_order", "value": planning_order},
)
# C-10, C-11, C-12: task-level checks
tasks = get_tasks_from_workplan(meta, body)
db_tasks = _api_get(api_base, "/tasks", {"workstream_id": ws_id})
@@ -626,6 +696,76 @@ def check_repo(api_base: str, repo_slug: str, repo_path_override: str | None = N
for t in db_tasks:
db_task_by_id[t["id"]] = t
existing_deps = _api_get(api_base, f"/workstreams/{ws_id}/dependencies") or []
existing_dep_keys = set()
if isinstance(existing_deps, list):
for dep in existing_deps:
if dep.get("from_workstream_id") != ws_id:
continue
rel = dep.get("relationship_type") or "blocks"
if dep.get("to_workstream_id"):
existing_dep_keys.add(("workstream", dep["to_workstream_id"], rel))
if dep.get("to_task_id"):
existing_dep_keys.add(("task", dep["to_task_id"], rel))
for target_wp_id in _as_list(meta.get("depends_on_workplans")):
target_ws_id = workplan_id_to_ws_id.get(target_wp_id)
if not target_ws_id:
report.add(
severity="WARN",
check_id="C-20",
message=f"Workplan dependency target '{target_wp_id}' is not linked to State Hub",
file_path=fname,
file_value=target_wp_id,
fixable=False,
)
continue
dep_key = ("workstream", target_ws_id, "blocks")
if dep_key not in existing_dep_keys:
report.add(
severity="WARN",
check_id="C-20",
message=f"Missing DB dependency edge: {ws_id[:8]}… depends on workplan {target_wp_id}",
file_path=fname,
db_id=ws_id,
file_value=target_wp_id,
fixable=True,
_fix_context={
"from_workstream_id": ws_id,
"to_workstream_id": target_ws_id,
"relationship_type": "blocks",
},
)
for target_task_id in _as_list(meta.get("depends_on_tasks")):
target_sh_id = task_file_id_to_sh_id.get(target_task_id)
if not target_sh_id:
report.add(
severity="WARN",
check_id="C-20",
message=f"Task dependency target '{target_task_id}' is not linked to State Hub",
file_path=fname,
file_value=target_task_id,
fixable=False,
)
continue
dep_key = ("task", target_sh_id, "starts_after")
if dep_key not in existing_dep_keys:
report.add(
severity="WARN",
check_id="C-20",
message=f"Missing DB dependency edge: {ws_id[:8]}… starts after task {target_task_id}",
file_path=fname,
db_id=ws_id,
file_value=target_task_id,
fixable=True,
_fix_context={
"from_workstream_id": ws_id,
"to_task_id": target_sh_id,
"relationship_type": "starts_after",
},
)
file_task_sh_ids: set[str] = set()
for task in tasks:
@@ -1180,7 +1320,7 @@ def fix_repo(
for issue in fixable:
ctx = issue._fix_context
try:
if issue.check_id in ("C-04", "C-05", "C-13"):
if issue.check_id in ("C-04", "C-05", "C-13", "C-19"):
ws_id = ctx["ws_id"]
result = _api_patch(api_base, f"/workstreams/{ws_id}",
{ctx["field"]: ctx["value"]})
@@ -1229,6 +1369,8 @@ def fix_repo(
"title": title or wp_id,
"status": status,
"owner": str(meta.get("owner", "")).strip() or None,
"planning_priority": str(meta.get("planning_priority", "")).strip() or None,
"planning_order": _as_int_or_none(meta.get("planning_order")),
})
if ws_data is None:
report.fixes_applied.append(
@@ -1284,6 +1426,25 @@ def fix_repo(
f"repo_id → {correct_repo_id[:8]}…"
)
elif issue.check_id == "C-20":
from_workstream_id = ctx["from_workstream_id"]
body = {
"to_workstream_id": ctx.get("to_workstream_id"),
"to_task_id": ctx.get("to_task_id"),
"relationship_type": ctx["relationship_type"],
}
result = _api_post(api_base, f"/workstreams/{from_workstream_id}/dependencies", body)
if result is not None and "_error" not in result:
target = ctx.get("to_workstream_id") or ctx.get("to_task_id")
report.fixes_applied.append(
f"C-20 fixed: dependency {from_workstream_id[:8]}… "
f"{ctx['relationship_type']} → {target[:8]}…"
)
elif result is not None:
report.fixes_applied.append(
f"C-20 FAILED: {result['_error']}"
)
elif issue.check_id == "C-10":
task_id = ctx["task_id"]
status = ctx["status"]