C-20 fixer: skip task-scoped edges when the hub lacks from_task_id

Guards a checker newer than the hub it talks to; the primary silently
drops unknown fields and would store a workplan-level edge that later
duplicates the task-scoped one.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 237582@bnt-lap001
Assistant-Session: f2b3d9f1-8fb9-4b9c-bc2b-837ec5dfc826

scripts/consistency_check.py

@@ -831,6 +831,31 @@ def _api_put(api_base: str, path: str, body: dict) -> Any:
         return {"_error": str(exc)}
 
 
+_FROM_TASK_ID_SUPPORT: dict[str, bool] = {}
+
+
+def _hub_accepts_from_task_id(api_base: str) -> bool:
+    """True when the hub's dependency-create schema declares from_task_id.
+
+    Cached per api_base. An unreadable schema counts as unsupported so the
+    fixer never writes a task-scoped edge into a hub that would drop the
+    attribution silently.
+    """
+    if api_base not in _FROM_TASK_ID_SUPPORT:
+        # Not via _api_get: it appends a trailing slash, which /openapi.json rejects.
+        try:
+            r = _api_client(api_base).get("/openapi.json")
+            spec = r.json() if r.status_code == 200 else {}
+        except Exception:
+            spec = {}
+        schemas = spec.get("components", {}).get("schemas", {}) if isinstance(spec, dict) else {}
+        _FROM_TASK_ID_SUPPORT[api_base] = any(
+            "from_task_id" in (schemas.get(name) or {}).get("properties", {})
+            for name in ("WorkplanDependencyCreate", "WorkstreamDependencyCreate")
+        )
+    return _FROM_TASK_ID_SUPPORT[api_base]
+
+
 def _api_post(api_base: str, path: str, body: dict) -> Any:
     if not _HAS_HTTPX:
         return {"_error": "httpx is not installed"}
@@ -3659,6 +3684,15 @@ def fix_repo(
                     "relationship_type": ctx["relationship_type"],
                 }
                 if ctx.get("from_task_id"):
+                    # A hub without migration e8f9a0b1c2d3 ignores the field
+                    # and would store a workplan-level edge that later
+                    # duplicates the task-scoped one. Skip until deployed.
+                    if not _hub_accepts_from_task_id(api_base):
+                        report.fixes_applied.append(
+                            f"C-20 SKIP: hub at {api_base} does not accept "
+                            f"from_task_id yet (task '{ctx.get('description', '')}')"
+                        )
+                        continue
                     body["from_task_id"] = ctx["from_task_id"]
                     body["description"] = ctx.get("description")
                 result = _api_post(api_base, f"/workplans/{from_workstream_id}/dependencies", body)

tests/test_consistency_check.py

@@ -2407,6 +2407,7 @@ class TestTaskWaitQualifiers:
             "consistency_check._api_patch",
             lambda _api_base, path, body: patches.append((path, body)) or {"ok": True},
         )
+        monkeypatch.setattr("consistency_check._hub_accepts_from_task_id", lambda _api_base: True)
         monkeypatch.setattr("consistency_check._detect_behind_remote", lambda _repo_path: False)
         monkeypatch.setattr("consistency_check._detect_ahead_of_remote", lambda _repo_path: 0)
         monkeypatch.setattr("consistency_check._git_commit_writeback", lambda *args, **kwargs: True)