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

Receive source-owned depends_on edges through Forge projection (STATE-WP-0094-T05)

Browse source 
Derive frontmatter and task-block depends_on edges from the pinned commit and
reconcile workplan_dependencies under an explicit ownership contract: only
forge:<repo> rows are created or removed, external relations are preserved,
missing/ambiguous/self targets are recorded and hold owned edges, failures
roll back, and repeat receipts are noop. Additive migration b3c4d5e6f7a8.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 671493@bnt-lap001
Assistant-Session: 60d7b11f-78f9-4731-9367-918511b2762d
This commit is contained in:
tegwick 2026-10-04 12:54:57 +02:00
parent b2af3c1c1d
commit fceb5f27f0
6 changed files with 637 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

301
api/services/forge_projection.py
View file

@ -138,6 +138,10 @@ class DerivedTask:
decision_id: str | None = None
blocked_on: str | None = None
wait_until: date | None = None
# Task-block ``depends_on`` targets (workplan or task record ids), in
# source order, deduplicated. Unqualified ``T##`` targets are qualified
# with the owning workplan.
depends_on: list[str] = field(default_factory=list)
@dataclass
@ -155,6 +159,10 @@ class DerivedWorkplan:
flavor_promotion_reason: str | None = None
flavor_promoted_from: str | None = None
historical_preservation: bool = False
# Frontmatter ``depends_on`` / ``depends_on_workplans`` / ``depends_on_tasks``
# split as C-20 splits them: task record ids vs workplan record ids.
depends_on_workplans: list[str] = field(default_factory=list)
depends_on_tasks: list[str] = field(default_factory=list)
@dataclass
@ -207,6 +215,8 @@ class DerivedProjection:
"relative_path": w.relative_path,
"archived": w.archived,
"historical_preservation": w.historical_preservation,
"depends_on_workplans": list(w.depends_on_workplans),
"depends_on_tasks": list(w.depends_on_tasks),
"tasks": [
{
"record_id": t.record_id,
@ -222,6 +232,7 @@ class DerivedProjection:
"decision_id": t.decision_id,
"blocked_on": t.blocked_on,
"wait_until": t.wait_until.isoformat() if t.wait_until else None,
"depends_on": list(t.depends_on),
}
for t in w.tasks
],
@ -353,6 +364,50 @@ def _source_needs_human(value: Any) -> bool:
raise ForgeDeriveError("Source needs_human must be boolean; no projection was applied")
_TASK_TARGET_RE = re.compile(r"-T\d{2,3}$")
_SHORT_TASK_TARGET_RE = re.compile(r"^T\d{2,3}$", re.IGNORECASE)
def _source_dependency_list(value: Any, field_name: str) -> list[str]:
"""Return declared dependency targets, refusing shapes C-20 cannot read.
A mapping or nested list is not a dependency declaration; dropping it would
silently lose an edge, so the whole derive is refused instead.
"""
if value is None or value == "":
return []
if isinstance(value, str):
items = value.split(",")
elif isinstance(value, list):
if any(isinstance(item, (dict, list)) for item in value):
raise ForgeDeriveError(f"Malformed source {field_name}; no projection was applied")
items = [str(item) for item in value if item is not None]
elif isinstance(value, dict):
raise ForgeDeriveError(f"Malformed source {field_name}; no projection was applied")
else:
items = [str(value)]
out: list[str] = []
for item in items:
target = item.strip().strip('"').strip("'").strip()
if target and target not in out:
out.append(target)
return out
def _split_dependency_targets(meta: dict, owner_record_id: str) -> tuple[list[str], list[str]]:
"""Split declared targets into (workplan ids, task ids) like C-20 does."""
declared = _source_dependency_list(meta.get("depends_on"), "depends_on")
workplans = [t for t in _source_dependency_list(meta.get("depends_on_workplans"), "depends_on_workplans")]
tasks = _source_dependency_list(meta.get("depends_on_tasks"), "depends_on_tasks")
for target in declared:
if _SHORT_TASK_TARGET_RE.match(target):
target = qualify_task_id(target, owner_record_id) or target
(tasks if _TASK_TARGET_RE.search(target) else workplans).append(target)
tasks = [qualify_task_id(t, owner_record_id) or t if _SHORT_TASK_TARGET_RE.match(t) else t for t in tasks]
dedupe = lambda items: list(dict.fromkeys(items)) # noqa: E731
return dedupe(workplans), dedupe(tasks)
def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
headings = [
(m.start(), len(m.group(1)), m.group(2).strip()) for m in _HEADING_RE.finditer(body)
@ -421,11 +476,17 @@ def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
decision_id=_source_wait_text(block.get("decision_id")),
blocked_on=_source_wait_text(block.get("blocked_on")),
wait_until=_source_wait_until(block.get("wait_until")),
depends_on=_task_dependency_targets(block, workplan_id),
)
)
return out
def _task_dependency_targets(block: dict, workplan_id: str) -> list[str]:
workplans, tasks = _split_dependency_targets(block, workplan_id)
return list(dict.fromkeys(workplans + tasks))
def _workplan_description(body: str) -> str | None:
"""Return bounded prose under ``## Goal`` when one is present."""
match = re.search(r"^##\s+Goal\s*$", body, re.MULTILINE | re.IGNORECASE)
@ -492,6 +553,7 @@ def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> Derive
rid = str(meta.get("id") or "").strip()
if not rid:
continue
frontmatter_deps = _split_dependency_targets(meta, rid)
proj.workplans.append(
DerivedWorkplan(
record_id=rid,
@ -510,6 +572,8 @@ def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> Derive
flavor=_source_flavor(meta.get("flavor")),
flavor_promotion_reason=_source_promotion_reason(meta.get("flavor_promotion_reason")),
flavor_promoted_from=_source_flavor(meta.get("flavor_promoted_from"), "flavor_promoted_from"),
depends_on_workplans=frontmatter_deps[0],
depends_on_tasks=frontmatter_deps[1],
)
)
proj.workplans.sort(key=lambda w: w.record_id)
@ -854,6 +918,15 @@ class ResetOutcome:
created_tasks: list[str] = field(default_factory=list)
updated_tasks: list[str] = field(default_factory=list)
cancelled_tasks: list[str] = field(default_factory=list)
# Source-owned dependency edges (STATE-WP-0094-T05).
created_edges: list[dict[str, Any]] = field(default_factory=list)
removed_edges: list[dict[str, Any]] = field(default_factory=list)
# Declared edges whose target is missing, ambiguous or invalid: recorded,
# never silently dropped; the from-side's owned edges are retained.
unresolved_edges: list[dict[str, Any]] = field(default_factory=list)
# Declared edges already present as externally authored rows: preserved
# and not adopted, so receiving can never remove them.
external_edges: list[dict[str, Any]] = field(default_factory=list)
notes: list[str] = field(default_factory=list)
def to_dict(self) -> dict[str, Any]:
@ -871,6 +944,10 @@ class ResetOutcome:
"created_tasks": len(self.created_tasks),
"updated_tasks": len(self.updated_tasks),
"cancelled_tasks": len(self.cancelled_tasks),
"created_edges": len(self.created_edges),
"removed_edges": len(self.removed_edges),
"unresolved_edges": len(self.unresolved_edges),
"external_edges": len(self.external_edges),
},
"created": self.created,
"updated": self.updated,
@ -879,6 +956,10 @@ class ResetOutcome:
"created_tasks": self.created_tasks,
"updated_tasks": self.updated_tasks,
"cancelled_tasks": self.cancelled_tasks,
"created_edges": self.created_edges,
"removed_edges": self.removed_edges,
"unresolved_edges": self.unresolved_edges,
"external_edges": self.external_edges,
"refused": self.refused,
"notes": self.notes,
}
@ -1483,8 +1564,11 @@ async def reset_repository_projection(
)
return outcome
received_rows: dict[str, Any] = {}
for key, w in want.items():
row = matched.get(key)
if row is not None:
received_rows[key] = row
if row is None:
row = Workplan(
id=uuid.UUID(w.uuid),
@ -1505,6 +1589,7 @@ async def reset_repository_projection(
)
session.add(row)
await session.flush()
received_rows[key] = row
for t in w.tasks:
# Safe only because nothing exists to mis-match against: this
# workplan is new to the hub.
@ -1584,6 +1669,8 @@ async def reset_repository_projection(
r.projection_retired_reason = RETIRE_REASON
r.slug = _tombstone_slug(r.slug or str(r.id), now)
await _reconcile_source_edges(session, repo, repo_slug, derived, want, received_rows, outcome)
if (
outcome.created
or outcome.updated
@ -1592,11 +1679,225 @@ async def reset_repository_projection(
or outcome.created_tasks
or outcome.updated_tasks
or outcome.cancelled_tasks
or outcome.created_edges
or outcome.removed_edges
):
outcome.status = "applied"
return outcome
# ---------------------------------------------------------------------------
# Source-owned dependency edges (STATE-WP-0094-T05)
# docs/forge-dependency-edge-receiving.md
# ---------------------------------------------------------------------------
EDGE_SOURCE_PREFIX = "forge:"
_CLOSED_TASK_STATUSES = {"done", "cancel"}
def _edge_workplan_closed(status: str | None) -> bool:
from api.workplan_status import CLOSED_WORKPLAN_STATUSES, normalize_workplan_status
try:
return normalize_workplan_status(status or "") in CLOSED_WORKPLAN_STATUSES
except ValueError:
return False
def _edge_task_closed(status: str | None) -> bool:
from api.task_status import normalize_task_status
try:
return normalize_task_status(status or "todo", default="todo") in _CLOSED_TASK_STATUSES
except ValueError:
return False
def edge_source_owner(repo_slug: str) -> str:
return f"{EDGE_SOURCE_PREFIX}{repo_slug}"
async def _resolve_edge_target(session: Any, record_id: str, cache: dict) -> tuple[str, Any]:
"""Resolve a declared target fleet-wide to ``("ok", uuid)`` or a reason.
Workplans match by ADR-007 derived UUID or record-id slug; tasks by derived
UUID or ``record_id``. Projection-retired workplans (and their tasks) are
not targets. More than one candidate is ambiguous: no winner is chosen.
"""
if record_id in cache:
return cache[record_id]
from sqlalchemy import func, or_, select
from api.models.task import Task
from api.models.workplan import Workplan
derived = uuid.UUID(derived_record_uuid(record_id))
if _TASK_TARGET_RE.search(record_id):
stmt = (
select(Task.id)
.join(Workplan, Workplan.id == Task.workplan_id)
.where(or_(Task.id == derived, Task.record_id == record_id))
.where(Workplan.projection_retired_at.is_(None))
)
else:
stmt = select(Workplan.id).where(
or_(Workplan.id == derived, func.lower(Workplan.slug) == record_id.lower())
).where(Workplan.projection_retired_at.is_(None))
ids = sorted({row[0] for row in (await session.execute(stmt)).all()}, key=str)
if not ids:
result: tuple[str, Any] = ("missing", None)
elif len(ids) > 1:
result = ("ambiguous", [str(i) for i in ids])
else:
result = ("ok", ids[0])
cache[record_id] = result
return result
async def _reconcile_source_edges(
session: Any,
repo: Any,
repo_slug: str,
derived: DerivedProjection,
want: dict[str, DerivedWorkplan],
received_rows: dict[str, Any],
outcome: ResetOutcome,
) -> None:
"""Reconcile ``workplan_dependencies`` rows owned by this repository's source.
Ownership: a row is source-owned iff ``source_owner == forge:<repo-slug>``
and its from-side belongs to this repository. Only those rows are created or
removed here. Rows with any other (or no) owner are externally authored and
preserved, even when they duplicate a declared edge.
"""
from sqlalchemy import select
from api.models.task import Task
from api.models.workplan import Workplan
from api.models.workplan_dependency import WorkplanDependency
owner = edge_source_owner(repo_slug)
await session.flush()
# (from_workplan_id, from_task_id, to_workplan_id, to_task_id, relationship)
desired: dict[tuple, dict[str, Any]] = {}
# From-sides with an unresolved declaration keep every owned edge: the
# unresolved target may be one of them (e.g. newly ambiguous).
held_sides: set[tuple] = set()
cache: dict[str, tuple[str, Any]] = {}
async def _declare(row: Any, w: DerivedWorkplan, from_task: Any, from_label: str, target: str) -> None:
side = (row.id, from_task.id if from_task is not None else None)
is_task = bool(_TASK_TARGET_RE.search(target))
state, value = await _resolve_edge_target(session, target, cache)
base = {"workplan": w.record_id, "from": from_label, "target": target}
if state == "ok" and ((not is_task and value == row.id) or (is_task and from_task is not None and value == from_task.id)):
state, value = "self", None
if state != "ok":
held_sides.add(side)
outcome.unresolved_edges.append({**base, "reason": state, **({"candidates": value} if value else {})})
return
rel = "starts_after" if is_task else "blocks"
key = (row.id, side[1], None if is_task else value, value if is_task else None, rel)
desired.setdefault(key, {**base, "relationship_type": rel})
for key, w in want.items():
row = received_rows.get(key)
if row is None or w.historical_preservation:
continue
for target in w.depends_on_workplans + w.depends_on_tasks:
await _declare(row, w, None, w.record_id, target)
# C-20 parity: closed workplans and closed tasks are history, not live
# ordering, so their task-block edges are not desired.
if _edge_workplan_closed(w.status):
continue
task_rows = None
for t in w.tasks:
if not t.depends_on or _edge_task_closed(t.status):
continue
if task_rows is None:
task_rows = {
(tr.record_id or "").strip().lower(): tr
for tr in (await session.execute(select(Task).where(Task.workplan_id == row.id))).scalars()
}
from_task = task_rows.get(t.record_id.strip().lower())
if from_task is None:
held_sides.add((row.id, "*"))
for target in t.depends_on:
outcome.unresolved_edges.append(
{"workplan": w.record_id, "from": t.record_id, "target": target,
"reason": "from-task-missing"}
)
continue
for target in t.depends_on:
await _declare(row, w, from_task, t.record_id, target)
repo_workplan_ids = list(
(await session.execute(select(Workplan.id).where(Workplan.repo_id == repo.id))).scalars()
)
existing = list(
(
await session.execute(
select(WorkplanDependency).where(
WorkplanDependency.from_workplan_id.in_(repo_workplan_ids)
)
)
).scalars()
) if repo_workplan_ids else []
by_key: dict[tuple, Any] = {
(e.from_workplan_id, e.from_task_id, e.to_workplan_id, e.to_task_id, e.relationship_type): e
for e in existing
}
def _edge_dict(edge_key: tuple, info: dict[str, Any] | None = None) -> dict[str, Any]:
from_wp, from_task, to_wp, to_task, rel = edge_key
return {
**(info or {}),
"from_workplan_id": str(from_wp),
"from_task_id": str(from_task) if from_task else None,
"to_workplan_id": str(to_wp) if to_wp else None,
"to_task_id": str(to_task) if to_task else None,
"relationship_type": rel,
}
for edge_key in sorted(desired, key=str):
info = desired[edge_key]
current = by_key.get(edge_key)
if current is not None:
if current.source_owner != owner:
outcome.external_edges.append(_edge_dict(edge_key, info))
continue
from_wp, from_task, to_wp, to_task, rel = edge_key
session.add(
WorkplanDependency(
from_workplan_id=from_wp,
from_task_id=from_task,
to_workplan_id=to_wp,
to_task_id=to_task,
relationship_type=rel,
description=f"{info['from']} depends_on {info['target']}",
source_owner=owner,
source_commit=derived.commit,
)
)
outcome.created_edges.append(_edge_dict(edge_key, info))
for edge_key, edge in sorted(by_key.items(), key=lambda kv: str(kv[0])):
if edge.source_owner != owner or edge_key in desired:
continue
if (edge.from_workplan_id, edge.from_task_id) in held_sides or (
edge.from_workplan_id, "*"
) in held_sides:
outcome.notes.append(
f"Retained source-owned edge {edge.id}: its from-side has an unresolved declaration."
)
continue
await session.delete(edge)
outcome.removed_edges.append(_edge_dict(edge_key, {"id": str(edge.id)}))
# Integrity failures surface here, inside the caller's transaction.
await session.flush()
# ---------------------------------------------------------------------------
# Fleet form (STATE-WP-0083-T04)
# ---------------------------------------------------------------------------