bin/migrate-task - main - veryviolet/coordination

bin/migrate-task
385 строк13 KB
violet
feat(R8): bin/task + bin/inbox + bin/stand + migrate; drop claim_filter
14 май 2026, 12:55
14 май 2026, 12:5589992ef
#!/usr/bin/env python3
"""bin/migrate-task — convert legacy .md task files to the strict .yaml format.

Legacy format (current): a markdown file with multiple YAML blocks separated
by `---`, optionally interleaved with free prose.
  * the first dict-shaped YAML block is the header (id, stream, kind, ...);
  * subsequent dict-shaped YAML blocks whose only top-level key is a known
    "<kind>_block" name are progress blocks;
  * everything else (prose, headings) becomes the task's `description` field.

New format: single YAML file with header fields at top level and an
ordered `blocks:` list. See bin/task for the exact shape.

Usage:
  migrate-task --file path/to/task.md          one file
  migrate-task --queue feature_dev             all active in a queue
  migrate-task --all                           every .md task in coordination/
                                               (active queues only by default;
                                               pass --include-terminal to also
                                               touch verified/, archive/, etc.)

Flags:
  --dry-run         report what would be migrated; write nothing
  --keep-md         leave .md alongside new .yaml (default: rename to .md.legacy)
  --force           overwrite existing .yaml if present

Best-effort: per-block fields are remapped where the canonical name
changed (written_by→by, written_at→at, closed_by→by, closed_at→at).
Unknown fields are preserved verbatim under their original key.
"""

from __future__ import annotations

import argparse
import datetime
import os
import re
import sys
import uuid
from pathlib import Path
from typing import Any

import yaml


def normalize_iso(value: Any) -> Any:
    """Coerce datetime / loose date strings to ISO-8601 with 'T'."""
    if isinstance(value, datetime.datetime):
        if value.tzinfo is None:
            return value.strftime("%Y-%m-%dT%H:%M:%SZ")
        return value.strftime("%Y-%m-%dT%H:%M:%S") + (
            "Z" if value.utcoffset() == datetime.timedelta(0) else value.strftime("%z")
        )
    if isinstance(value, datetime.date):
        return value.strftime("%Y-%m-%d") + "T00:00:00Z"
    if isinstance(value, str) and re.match(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}", value):
        return value.replace(" ", "T", 1)
    return value


def walk_normalize(obj: Any) -> Any:
    if isinstance(obj, dict):
        return {k: walk_normalize(v) if not _is_ts_key(k) else normalize_iso(v)
                for k, v in obj.items()}
    if isinstance(obj, list):
        return [walk_normalize(x) for x in obj]
    return obj


def _is_ts_key(k: str) -> bool:
    if not isinstance(k, str):
        return False
    return k in ("at", "opened_at", "answered_at", "sent_at",
                 "closed_at", "written_at", "reviewed_at", "intent_at",
                 "gate_check_at")

# block-name → canonical kind
BLOCK_NAME_TO_KIND: dict[str, str] = {
    "triage_block": "triage",
    "triage": "triage",
    "plan_block": "plan",
    "plan": "plan",
    "implementation_block": "implementation",
    "impl_block": "implementation",
    "implementation": "implementation",
    "tests_block": "tests",
    "tests": "tests",
    "reader_block": "reader_review",
    "reader_review": "reader_review",
    "review_block": "review",
    "review": "review",
    "blocked_block": "blocked",
    "blocked": "blocked",
    "stand_result_block": "stand_result",
    "stand_result": "stand_result",
    "session_iteration": "session_iteration",
}

# legacy field name → canonical name
FIELD_REMAP: dict[str, str] = {
    "written_by": "by",
    "written_at": "at",
    "closed_by": "by",
    "closed_at": "at",
    "reviewed_by": "by",
    "reviewed_at": "at",
}


# Active product queues (default scope for --all).
ACTIVE_QUEUES = (
    "feature_inbox", "feature_plan",
    "feature_dev", "feature_ui_dev", "feature_docs",
    "feature_test", "feature_docs_review",
    "feature_review", "feature_blocked",
    "user_feedback", "review_sessions",
    "stand_requests", "stand_wip",
)

TERMINAL_QUEUES = ("verified", "archive", "stand_done", "bot_archive", "bot_verified")


def die(msg: str, code: int = 1) -> None:
    print(f"migrate-task: {msg}", file=sys.stderr)
    sys.exit(code)


def find_coord_dir() -> Path:
    env_dir = os.environ.get("COORD_PROJECT_DIR")
    if env_dir:
        p = Path(env_dir) / "coordination"
        if p.is_dir():
            return p
    cur = Path.cwd()
    for p in [cur, *cur.parents]:
        c = p / "coordination"
        if c.is_dir():
            return c
    die("coordination/ not found")
    raise SystemExit


def split_md_chunks(text: str) -> list[str]:
    """Split on lines that are exactly '---', keep contents between."""
    return re.split(r"^---\s*$", text, flags=re.MULTILINE)


def chunk_yaml(chunk: str) -> Any:
    """Try parse chunk as YAML; return value or None."""
    chunk = chunk.strip()
    if not chunk:
        return None
    try:
        return yaml.safe_load(chunk)
    except yaml.YAMLError:
        return None


def remap_block_fields(block: dict[str, Any]) -> dict[str, Any]:
    out: dict[str, Any] = {}
    for k, v in block.items():
        new_k = FIELD_REMAP.get(k, k)
        if new_k in out and out[new_k] != v:
            # if both written_by and closed_by present, keep the later one
            out[new_k] = v
        else:
            out[new_k] = v
    return out


def migrate_one(path: Path, dry_run: bool, force: bool, keep_md: bool) -> tuple[str, str | None]:
    """Return (status, message). status: 'migrated' | 'skipped' | 'error'."""
    if path.suffix != ".md":
        return ("skipped", "not a .md file")
    yaml_path = path.with_suffix(".yaml")
    if yaml_path.exists() and not force:
        return ("skipped", f"{yaml_path.name} already exists (use --force)")

    text = path.read_text(encoding="utf-8")
    chunks = split_md_chunks(text)

    header: dict[str, Any] = {}
    blocks: list[dict[str, Any]] = []
    prose_parts: list[str] = []

    for chunk in chunks:
        data = chunk_yaml(chunk)
        if data is None:
            # plain prose
            s = chunk.strip()
            if s:
                prose_parts.append(s)
            continue
        if not isinstance(data, dict):
            # weird (list at top level) — skip but warn
            continue
        # is this header? heuristic: first dict with 'id' field
        if not header and "id" in data:
            header = dict(data)
            continue
        # is it a known block? heuristic: dict has exactly one key that
        # matches a known block name, OR has 'kind' field directly.
        if "kind" in data and data["kind"] in BLOCK_NAME_TO_KIND.values():
            blocks.append(remap_block_fields(dict(data)))
            continue
        if len(data) == 1:
            (k, v), = data.items()
            if k in BLOCK_NAME_TO_KIND and isinstance(v, dict):
                block = dict(v)
                block["kind"] = BLOCK_NAME_TO_KIND[k]
                blocks.append(remap_block_fields(block))
                continue
        # check if dict-as-whole-block-payload: keys include 'written_by'
        # or 'closed_by' etc — treat as bare block (kind unknown). Skip
        # rather than guess.
        if any(k in data for k in ("written_by", "closed_by", "reviewed_by")):
            # block without explicit kind tag — try inferring
            inferred = infer_block_kind(data)
            if inferred:
                block = dict(data)
                block["kind"] = inferred
                blocks.append(remap_block_fields(block))
                continue
        # otherwise, treat as additional header data (merge into header)
        if header:
            for k, v in data.items():
                header.setdefault(k, v)

    if not header:
        return ("error", "no header (first YAML block with id) found")
    if not header.get("id"):
        return ("error", "header missing 'id'")

    # build new doc
    new_doc: dict[str, Any] = {}
    # canonical order of header fields
    for k in ("id", "stream", "kind", "scope", "title", "reporter",
              "opened_at", "priority", "assignee_role",
              "request_type", "target", "evidence_for",
              "mode", "target_functionality", "scenarios", "stand_target",
              "related"):
        if k in header:
            new_doc[k] = header[k]
    # copy any other header fields verbatim (not already moved)
    for k, v in header.items():
        if k not in new_doc:
            new_doc[k] = v

    if prose_parts:
        new_doc["description"] = "\n\n".join(prose_parts)

    new_doc["blocks"] = ensure_block_meta(blocks)

    # normalize legacy date/datetime values to ISO strings
    new_doc = walk_normalize(new_doc)

    if dry_run:
        return ("migrated", f"would write {yaml_path.name} "
                            f"({len(blocks)} block(s), prose: {bool(prose_parts)})")

    write_yaml_atomic(yaml_path, new_doc)

    if keep_md:
        return ("migrated", f"wrote {yaml_path.name} (kept {path.name})")
    legacy = path.with_suffix(".md.legacy")
    try:
        os.rename(path, legacy)
    except OSError as exc:
        return ("error", f"rename {path.name} → .legacy failed: {exc}")
    return ("migrated", f"wrote {yaml_path.name}, renamed {path.name} → {legacy.name}")


def infer_block_kind(data: dict[str, Any]) -> str | None:
    """Heuristic: block payload without explicit kind tag."""
    has = lambda *keys: any(k in data for k in keys)
    if has("plan_kind", "assignee_role") and has("ready_for_implementation"):
        return "plan"
    if has("ready_for_test", "files", "files_touched"):
        return "implementation"
    if has("test_result", "test_command", "gate_check_result"):
        return "tests"
    if has("docs_checked", "stand_checked", "command_or_flow"):
        return "reader_review"
    if has("review_block", "approved") or data.get("outcome") in ("approved", "changes_requested"):
        return "review"
    if has("dependencies", "resume_to"):
        return "blocked"
    if has("stand_status", "result") and has("profile"):
        return "stand_result"
    return None


def ensure_block_meta(blocks: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Make sure every block has 'by' and 'at'; leave missing fields as None
    but keep the block (legacy files are messy)."""
    out: list[dict[str, Any]] = []
    for b in blocks:
        nb = dict(b)
        # 'kind' must be at front for readability
        ordered: dict[str, Any] = {}
        for k in ("kind", "by", "at"):
            if k in nb:
                ordered[k] = nb.pop(k)
        ordered.update(nb)
        out.append(ordered)
    return out


def write_yaml_atomic(path: Path, data: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + f".tmp.{uuid.uuid4().hex[:8]}")
    try:
        with tmp.open("w", encoding="utf-8") as f:
            yaml.safe_dump(data, f, sort_keys=False, allow_unicode=True,
                           default_flow_style=False)
            f.flush()
            os.fsync(f.fileno())
        os.rename(tmp, path)
    finally:
        try:
            tmp.unlink()
        except OSError:
            pass


def collect_targets(args: argparse.Namespace) -> list[Path]:
    if args.file:
        return [Path(args.file).resolve()]
    coord = find_coord_dir()
    queues: list[str] = []
    if args.queue:
        queues = [args.queue]
    elif args.all:
        queues = list(ACTIVE_QUEUES)
        if args.include_terminal:
            queues += list(TERMINAL_QUEUES)
    else:
        die("specify --file FILE, --queue QUEUE, or --all")
    out: list[Path] = []
    for q in queues:
        qd = coord / q
        if not qd.is_dir():
            continue
        for f in qd.glob("*.md"):
            if f.name == "_TEMPLATE.md":
                continue
            out.append(f)
    return out


def main(argv: list[str]) -> int:
    ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    g = ap.add_mutually_exclusive_group()
    g.add_argument("--file", help="single .md file to migrate")
    g.add_argument("--queue", help="migrate all .md in this queue")
    g.add_argument("--all", action="store_true",
                   help="migrate every .md in active queues")
    ap.add_argument("--include-terminal", action="store_true",
                    help="with --all, also include verified/archive/stand_done/...")
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--keep-md", action="store_true",
                    help="don't rename original .md to .md.legacy")
    ap.add_argument("--force", action="store_true",
                    help="overwrite existing .yaml")
    args = ap.parse_args(argv)

    targets = collect_targets(args)
    if not targets:
        print("migrate-task: no .md tasks found")
        return 0

    stats = {"migrated": 0, "skipped": 0, "error": 0}
    for path in targets:
        status, msg = migrate_one(path, args.dry_run, args.force, args.keep_md)
        stats[status] = stats.get(status, 0) + 1
        if status != "skipped" or args.dry_run:
            tag = {"migrated": "OK", "skipped": "--", "error": "ERR"}[status]
            print(f"  [{tag}] {path.name}: {msg}")
    print(f"\nmigrated: {stats['migrated']}  "
          f"skipped: {stats['skipped']}  errors: {stats['error']}")
    return 0 if stats["error"] == 0 else 2


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))