coulomb/state-hub
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

CUST-WP-0075: reconcile owner waits and decision routing

Browse source 
Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a10391-6355-7b41-b0fa-3e14e874f091
This commit is contained in:
tegwick 2026-10-04 00:48:22 +02:00
parent d61ba327fa
commit 831ae0e598
3 changed files with 233 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

73
scripts/consistency_check.py
View file

@ -42,7 +42,7 @@ Checks:
C-36 work-record-flavor-unknown WARN No flavor is set but not in the closed list
C-37 residual-provenance-missing WARN No flavor: residual without origin/origin_ref
C-38 task-flavor-drift WARN Yes task flavor differs between file and DB (file wins)
C-39 task-wait-unqualified WARN No wait task has none of task-block depends_on, needs_human: true, blocked_on: message-from:<agent>, wait_until: YYYY-MM-DD (CUST-WP-0074 rules 1/6)
C-39 task-wait-unqualified WARN No unrouted human wait (open non-residual) or wait task has none of task-block depends_on, needs_human: true, blocked_on: message-from:<agent>, wait_until: YYYY-MM-DD (CUST-WP-0074 rules 1/6)
C-40 task-wait-blocker-satisfied WARN No wait task whose depends_on targets are all terminal, whose decision_id is resolved, or whose wait_until has passed (CUST-WP-0074 rules 3/4)
C-41 task-wait-qualifier-drift WARN Yes needs_human/blocking_reason/decision_id/blocked_on/wait_until differ between file and DB (file wins)
C-42 workplan-blocked-no-wait WARN No blocked workplan has no task in status wait (CUST-WP-0074-T06)
@ -2082,6 +2082,21 @@ def check_repo(
file_value=t_status,
fixable=False,
)
# CUST-WP-0075 P1: human gates belong in the decision queue.
if (
wait_fields["needs_human"] and not wait_fields["decision_id"]
and not is_residual_flavor(meta.get("flavor"))
):
report.add(
severity="WARN", check_id="C-39",
message=(
f"Human wait task '{t_id}' in workplan '{meta.get('id', fname)}' "
"has needs_human: true without decision_id — unrouted human gate"
),
file_path=f"{fname}#{t_id}",
file_value="needs_human: true",
fixable=False,
)
# C-40: rule 3 for a wait window — the date has passed (UTC today).
if wait_until and _date.fromisoformat(wait_until) < datetime.now(timezone.utc).date():
report.add(
@ -2814,6 +2829,61 @@ def _check_blocked_unblock_sweep(
)
# Task-level returns can clear a wait even after somebody read the reply.
# Preserve the legacy workplan check above; task findings carry stable IDs
# for an owner runner to deduplicate without changing task status here.
messages_by_agent: dict[str, list[dict]] = {}
for wp_file, meta, body in workplan_infos:
if (
normalise_workstream_status(str(meta.get("status", ""))) in CLOSED_WORKSTREAM_STATUSES
or is_residual_flavor(meta.get("flavor"))
):
continue
for task in get_tasks_from_workplan(meta, body):
if normalise_task_status(task.get("status", "todo")) != "wait":
continue
counterpart = parse_blocked_on(str(task.get("blocked_on") or ""))
if not counterpart:
continue
if counterpart not in messages_by_agent:
rows = _api_get(api_base, "/messages", {
"to_agent": repo_slug, "from_agent": counterpart,
"unread_only": False, "limit": 200,
}) or []
messages_by_agent[counterpart] = rows if isinstance(rows, list) else []
# Dated notes are the available acknowledgement watermark. Date-only
# notes cover the whole UTC day, avoiding same-day re-notification.
note_dates = []
for value in re.findall(r"(?m)^\s*(?:[-*]\s+)?(?:\*\*)?(\d{4}-\d{2}-\d{2})\b", str(task.get("description") or "")):
parsed = _parse_message_timestamp(value)
if parsed:
note_dates.append(parsed + timedelta(days=1))
explicit_updated = _parse_message_timestamp(task.get("updated"))
cutoff = max(note_dates + ([explicit_updated] if explicit_updated else []), default=None)
hits = []
for message in messages_by_agent[counterpart]:
if message.get("from_agent") != counterpart or message.get("archived_at"):
continue
created = _parse_message_timestamp(message.get("created_at"))
if created and (created >= cutoff if cutoff else not message.get("read_at")):
hits.append(message)
if not hits:
continue
hits.sort(key=lambda message: str(message.get("created_at", "")))
task_id = str(task.get("id", "")).strip()
message_ids = [str(message["id"]) for message in hits if message.get("id")]
report.add(
severity="WARN", check_id="C-25",
message=(f"Owner-return task '{task_id}' waiting on {counterpart} has "
f"{len(hits)} reply/replies after its last dated note "
f"(or unread when undated) — owner action needed: {', '.join(message_ids)}"),
file_path=f"{workplan_display_path(repo_dir, wp_file)}#{task_id}",
file_value=str(task.get("blocked_on")),
fixable=False,
_fix_context={"task_record_id": task_id, "message_ids": message_ids},
)
def _check_inbox_hygiene(api_base: str, repo_slug: str, report: ConsistencyReport) -> None:
hygiene = collect_inbox_hygiene(api_base, repo_slug)
if hygiene["stale_unread_count"]:
@ -4254,6 +4324,7 @@ def report_to_dict(report: ConsistencyReport) -> dict:
"file_value": i.file_value,
"db_value": i.db_value,
"fixable": i.fixable,
**({"owner_return": i._fix_context} if i.check_id == "C-25" and i._fix_context.get("task_record_id") else {}),
}
for i in report.issues
],