75 lines
2.6 KiB
Python
75 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Immutable SCM lineage carried by supervised Kanban follow-up cards."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Lineage:
|
|
"""Trusted root, project, branch, and pull request for one change chain."""
|
|
|
|
root_task_id: str
|
|
branch: str
|
|
pull_request: str
|
|
project: str = ""
|
|
base_branch: str = ""
|
|
|
|
def stamp_fields(self) -> dict[str, str]:
|
|
"""Return immutable fields safe to embed in a supervisor stamp."""
|
|
return {"root_task_id": self.root_task_id, "branch": self.branch,
|
|
"pull_request": self.pull_request, "project": self.project,
|
|
"base_branch": self.base_branch}
|
|
|
|
|
|
def _mapping(value: Any) -> dict[str, Any]:
|
|
if isinstance(value, dict):
|
|
return value
|
|
if isinstance(value, str):
|
|
try:
|
|
parsed = json.loads(value)
|
|
except (TypeError, ValueError):
|
|
return {}
|
|
return parsed if isinstance(parsed, dict) else {}
|
|
return {}
|
|
|
|
|
|
def _text(source: dict[str, Any], *keys: str) -> str:
|
|
for key in keys:
|
|
value = source.get(key)
|
|
if isinstance(value, str) and value.strip():
|
|
return value.strip()
|
|
return ""
|
|
|
|
|
|
def initial(task: Any) -> Lineage | None:
|
|
"""Read only coordinator-written assignment metadata, never worker prose."""
|
|
raw = task.get("metadata") if isinstance(task, dict) else getattr(task, "metadata", None)
|
|
meta = _mapping(raw)
|
|
sources = (_mapping(meta.get("supervisor_lineage")), _mapping(meta.get("assignment")))
|
|
task_id = str(task.get("id") if isinstance(task, dict) else getattr(task, "id", "") or "")
|
|
for source in sources:
|
|
branch = _text(source, "branch", "branch_name", "head_branch")
|
|
pull = _text(source, "pull_request", "pr_url", "pr", "merge_request")
|
|
base = _text(source, "base_branch")
|
|
if branch and pull and base:
|
|
root = _text(source, "root_task_id") or task_id
|
|
if root:
|
|
return Lineage(root, branch, pull, _text(source, "project", "repository", "repo"), base)
|
|
return None
|
|
|
|
|
|
def from_stamp(value: Any) -> Lineage | None:
|
|
"""Validate immutable continuation fields written by the supervisor itself."""
|
|
source = _mapping(value)
|
|
root = _text(source, "root_task_id")
|
|
branch = _text(source, "branch")
|
|
pull = _text(source, "pull_request")
|
|
base = _text(source, "base_branch")
|
|
if not root or not branch or not pull or not base or _text(source, "root") != root:
|
|
return None
|
|
return Lineage(root, branch, pull, _text(source, "project"), base)
|