#!/usr/bin/env python3
# GENERATED FROM SPEC — DO NOT EDIT DIRECTLY
# Source: openspec/specs/local-handoff-memory/spec.md
"""Validate and import handoff graph sidecars and facts into a local Neo4j graph.

Subcommands:
  validate  offline checks only (no Neo4j, no driver needed)
  import    one-transaction upsert + per-handoff reconciliation
  rebuild   one-transaction full replacement of the dedicated namespace

Every input is read once, validated completely (schema selected by version plus
whole-input consistency) and frozen in memory before any driver is created.
What is written comes from the declarative graph model (E24,
references/graph-model.yaml of the handoff skill): E21 v1 / E23 v2 sidecars and E24
facts files all become model facts; node labels, relationship types, write
queries, reconciliation sets and parentage checks are derived from the model and
never from input content. Source Markdown, sidecars, facts and the E21 project
identity file are never written. Only private queue state is written. New
automatic commands also support the local runtime credential file.
"""

from __future__ import annotations

import argparse
import contextlib
import fcntl
import re
import subprocess
import tempfile
import hashlib
import importlib.util
import json
import os
import sys
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Callable, Iterable
from urllib.parse import urlsplit


# Skill files are resolved from this script's own location, never from the
# current directory or a repository root, so a copy of the skill works anywhere.
SCRIPTS_DIR = Path(__file__).resolve().parent
E21_EXPORTER = SCRIPTS_DIR / "graph_export.py"
MODEL_ENGINE = SCRIPTS_DIR / "handoff_graph_model.py"

SIDECAR_GLOB = "HANDOFF-*.graph.json"
SIDECAR_SUFFIX = ".graph.json"
FACTS_GLOB = "HANDOFF-*.facts.json"
FACTS_SUFFIX = ".facts.json"
SUPPORTED_VERSIONS = (1, 2)
DEFAULT_URI = "bolt://127.0.0.1:7687"
DEFAULT_USER = "neo4j"
DEFAULT_DATABASE = "neo4j"
LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1"}
ALLOWED_SCHEMES = {"bolt", "neo4j"}

EXIT_OK = 0
EXIT_INPUT = 1
EXIT_DATABASE = 3


# --------------------------------------------------------------------------- errors

class ImportFailure(Exception):
    """Base error; messages never contain source content or credentials."""


class InputError(ImportFailure):
    def __init__(self, category: str, path: str | None = None) -> None:
        self.category = category
        self.path = path
        where = f"{path}: " if path else ""
        super().__init__(f"{where}{category}")


class ConfigError(ImportFailure):
    pass


class ConflictError(ImportFailure):
    pass


class BoundaryError(ImportFailure):
    pass


# --------------------------------------------------------------------------- E21 helpers

def _load_e21() -> Any:
    spec = importlib.util.spec_from_file_location("_e21_graph_export", E21_EXPORTER)
    if spec is None or spec.loader is None:
        raise InputError("E21 exporter unavailable")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


_E21_CACHE: dict[str, Any] = {}


def _e21_module() -> Any:
    if "module" not in _E21_CACHE:
        _E21_CACHE["module"] = _load_e21()
    return _E21_CACHE["module"]



def _load_engine() -> Any:
    name = "handoff_graph_model"
    if name not in sys.modules:
        spec = importlib.util.spec_from_file_location(name, MODEL_ENGINE)
        module = importlib.util.module_from_spec(spec)
        sys.modules[name] = module
        spec.loader.exec_module(module)
    return sys.modules[name]


engine = _load_engine()
context_id = engine.context_id


def _handoff_number(filename: str) -> int:
    return int(filename[len("HANDOFF-"): -len(".md")])


# --------------------------------------------------------------------------- model-derived schema

_MERGE_NODE = "UNWIND $rows AS row MERGE (n:HandoffMemory {namespace: $ns, id: row.id}) "
_MERGE_EDGE_HEAD = (
    "UNWIND $rows AS row "
    "MATCH (a:HandoffMemory {namespace: $ns, id: row.start}) "
    "MATCH (b:HandoffMemory {namespace: $ns, id: row.end}) "
)


class Schema:
    """Everything the importer writes, derived from one validated model.

    Labels and relationship types reach Cypher text only from here, i.e. only
    after the model validator accepted them; input content never does.
    """

    def __init__(self, model: Any) -> None:
        self.model = model
        self.namespace = model.namespace
        self.kind_labels: dict[str, tuple[str, ...]] = {name: node.all_labels for name, node in model.nodes.items()}
        self.upsert: dict[str, str] = {
            name: _MERGE_NODE + ("SET n += row.props SET n:" if node.enrich else "SET n = row.props SET n:") + ":".join(node.labels)
            for name, node in model.nodes.items()
        }
        self.merge_edge: dict[str, str] = {
            rel: _MERGE_EDGE_HEAD + "MERGE (a)-[r:" + rel + " {namespace: $ns}]->(b) SET r = row.props SET r.namespace = $ns"
            for rel in model.relationships
        }
        self.node_order = list(model.nodes)
        self.edge_order = list(model.relationships)
        self.containment = list(model.containment_rel_types)
        self.metadata_rels = list(model.handoff_owned_rel_types)
        self.owned = sorted(model.relationships)
        self.record_rels = list(model.record_rel_types)
        self.record_types = set(model.record_node_types)
        self.record_out_rels = list(model.record_out_rel_types)
        self.deterministic_record_out_rels = list(model.deterministic_record_out_rel_types)
        self.key_labels = list(model.key_labels)
        self.handoff_type = model.handoff_type
        self.handoff_labels = list(model.nodes[model.handoff_type].labels)
        self.single_parent = list(model.single_parent_rel_types)
        self.max_out = {name: rel.max_out for name, rel in model.relationships.items() if rel.max_out is not None}


def load_schema(path: str | Path | None = None) -> Schema:
    return Schema(engine.load_model(path))


SCHEMA = load_schema()
NAMESPACE = SCHEMA.namespace
KIND_LABELS = SCHEMA.kind_labels
Q_UPSERT = SCHEMA.upsert
Q_MERGE_EDGE = SCHEMA.merge_edge
CONTAINMENT_REL_TYPES = SCHEMA.containment
METADATA_REL_TYPES = SCHEMA.metadata_rels
OWNED_REL_TYPES = SCHEMA.owned


# --------------------------------------------------------------------------- frozen payload

@dataclass(frozen=True)
class Node:
    id: str
    kind: str
    parent: str | None
    props: tuple[tuple[str, Any], ...]

    def props_dict(self) -> dict[str, Any]:
        return dict(self.props)


@dataclass(frozen=True)
class Edge:
    type: str
    start: str
    end: str
    props: tuple[tuple[str, Any], ...] = ()


@dataclass(frozen=True)
class Payload:
    sources: tuple[str, ...]
    nodes: tuple[Node, ...]
    edges: tuple[Edge, ...]
    handoff_records: tuple[tuple[str, tuple[str, ...]], ...]
    handoff_metadata: tuple[tuple[str, tuple[str, ...]], ...] = ()
    external_handoffs: tuple[tuple[str, str], ...] = ()
    child_of: tuple[tuple[str, str], ...] = ()
    single_parent: tuple[tuple[str, tuple[tuple[str, str], ...]], ...] = ()
    external_nodes: tuple[tuple[str, str], ...] = ()
    record_edges: tuple[tuple[str, tuple[str, ...], tuple[str, ...]], ...] = ()
    schema: Any = field(default=None, compare=False, repr=False)

    @property
    def active_schema(self) -> Schema:
        return self.schema or SCHEMA

    def digest(self) -> str:
        return "sha256:" + hashlib.sha256(canonical_payload(self)).hexdigest()

    def nodes_of(self, kind: str) -> list[Node]:
        return [node for node in self.nodes if node.kind == kind]

    def edges_of(self, rel: str) -> list[Edge]:
        return [edge for edge in self.edges if edge.type == rel]


def canonical_payload(payload: Payload) -> bytes:
    schema = payload.active_schema
    data = {
        "namespace": schema.namespace,
        "nodes": [
            {"id": n.id, "kind": n.kind, "labels": list(schema.kind_labels[n.kind]), "parent": n.parent, "props": dict(n.props)}
            for n in payload.nodes
        ],
        "edges": [[e.type, e.start, e.end, dict(e.props)] if e.props else [e.type, e.start, e.end] for e in payload.edges],
    }
    return json.dumps(data, sort_keys=True, ensure_ascii=False).encode("utf-8")


# --------------------------------------------------------------------------- discovery and strict loading

def discover(paths: Iterable[str | Path], facts: bool = False) -> list[Path]:
    """Resolve explicit files and direct HANDOFF-*.graph.json (or, with facts, *.facts.json) directory children."""
    glob = FACTS_GLOB if facts else SIDECAR_GLOB
    found: set[Path] = set()
    for raw in paths:
        path = Path(raw).expanduser()
        if path.is_dir():
            for child in path.glob(glob):
                if child.is_file():
                    found.add(child.resolve())
        elif path.is_file():
            found.add(path.resolve())
        else:
            raise InputError("input path does not exist", str(raw))
    if not found:
        raise InputError(f"no {glob} files discovered")
    return sorted(found)


def _no_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate key")
        result[key] = value
    return result


def _reject_constant(_: str) -> Any:
    raise ValueError("non-standard JSON constant")


def load_strict(path: Path) -> tuple[bytes, Any]:
    try:
        raw = path.read_bytes()
    except OSError as exc:
        raise InputError("unreadable input", str(path)) from exc
    try:
        text = raw.decode("utf-8", errors="strict")
    except UnicodeDecodeError as exc:
        raise InputError("malformed UTF-8", str(path)) from exc
    try:
        document = json.loads(text, object_pairs_hook=_no_duplicate_keys, parse_constant=_reject_constant)
    except ValueError as exc:
        if "duplicate key" in str(exc):
            raise InputError("duplicate JSON object key", str(path)) from exc
        raise InputError("malformed JSON", str(path)) from exc
    return raw, document


# --------------------------------------------------------------------------- validation

def _validate_document(path: Path, document: Any) -> None:
    module = _e21_module()
    if isinstance(document, dict) and "schema_version" in document and document["schema_version"] not in SUPPORTED_VERSIONS:
        raise InputError("unsupported schema version", str(path))
    try:
        module.validate_document(document)
    except module.ExportError as exc:
        raise InputError(f"schema mismatch ({exc})", str(path)) from exc
    filename = document["handoff"]["filename"]
    expected_name = filename[: -len(".md")] + SIDECAR_SUFFIX
    if path.name != expected_name:
        raise InputError("sidecar filename does not match handoff filename", str(path))
    seen: set[str] = set()
    for record in document["records"]:
        if record["id"] in seen:
            raise InputError("repeated record ID within document", str(path))
        seen.add(record["id"])
        if record["provenance"]["file"] != filename:
            raise InputError("record provenance file differs from handoff filename", str(path))
        if document["schema_version"] >= 2 and record["provenance"]["directory"] != document["source"]["directory"]:
            raise InputError("record provenance directory differs from the source folder", str(path))
    if document["schema_version"] >= 2:
        _validate_v2(path, document, module)


def _validate_v2(path: Path, document: dict[str, Any], module: Any) -> None:
    filename = document["handoff"]["filename"]
    metadata = document["metadata"]
    project = uuid.UUID(document["project"]["id"])
    if document["source"]["file"] != filename or metadata["handoff"] + ".md" != filename:
        raise InputError("metadata or source file differs from handoff filename", str(path))
    if document["project"]["project_id"] != metadata["project_id"]:
        raise InputError("canonical project id differs from metadata", str(path))
    if document["handoff"]["id"] != str(uuid.uuid5(project, f"handoff:{filename}")) or document["session"]["id"] != str(uuid.uuid5(project, f"session:{filename}")):
        raise InputError("handoff identity is not derived from its technical project identity", str(path))
    try:
        module._check_types(metadata, document["metadata_profile"])
    except module.ExportError as exc:
        raise InputError(f"invalid metadata ({exc})", str(path)) from exc
    if metadata["continues"] is not None and _handoff_number(metadata["continues"] + ".md") >= _handoff_number(filename):
        raise InputError("continuation must reference an earlier handoff", str(path))
    if metadata["parent_session"] is not None and metadata["agent_session"] is None:
        raise InputError("parent_session requires agent_session", str(path))

def _canonical_document(document: dict[str, Any]) -> bytes:
    return json.dumps(document, sort_keys=True, ensure_ascii=False).encode("utf-8")


def _register(identities: dict[str, tuple[str, str | None, Any]], ident: str, kind: str, parent: str | None, definition: Any, path: Path) -> None:
    current = (kind, parent, definition)
    previous = identities.get(ident)
    if previous is None:
        identities[ident] = current
        return
    if previous[0] != kind:
        raise InputError("identity reused for a different entity kind", str(path))
    if previous[1] != parent:
        raise InputError("identity reused with conflicting parentage", str(path))
    if previous[2] != definition:
        raise InputError("identity reused with a conflicting definition", str(path))


def _facts_from_file(path: Path, data: Any, schema: Schema) -> dict[str, Any]:
    if not path.name.endswith(FACTS_SUFFIX):
        raise InputError("facts file name must end with .facts.json", str(path))
    try:
        engine.validate_facts(data, schema.model)
    except engine.ExtractError as exc:
        raise InputError(f"invalid facts ({exc})", str(path)) from exc
    filename = data["handoff"]["filename"]
    if path.name != filename[: -len(".md")] + FACTS_SUFFIX:
        raise InputError("facts filename does not match handoff filename", str(path))
    return data


def _check_single_parent(parents: dict[str, str], stored: dict[str, str], rel: str = "CHILD_OF") -> None:
    """Reject conflicting parents and cycles of a single-parent relationship across batch and stored graph."""
    combined = dict(stored)
    for child, parent in parents.items():
        if combined.setdefault(child, parent) != parent:
            raise ConflictError(f"stored {rel} source has a conflicting parent")
    for start in combined:
        seen = {start}
        node = combined.get(start)
        while node is not None:
            if node in seen:
                raise InputError(f"{rel} parentage forms a cycle")
            seen.add(node)
            node = combined.get(node)


_check_session_ancestry = _check_single_parent


def build_payload(paths: Iterable[str | Path], schema: Schema | None = None, facts: bool = False,
                  frozen: dict[Path, Any] | None = None) -> Payload:
    """Read each discovered input once, validate everything, turn it into model facts, freeze the result."""
    schema = schema or SCHEMA
    model = schema.model
    files = discover(paths, facts)
    entries: dict[str, tuple[Path, dict[str, Any], bytes]] = {}
    for path in files:
        document = frozen[path] if frozen is not None else load_strict(path)[1]
        if path.name.endswith(FACTS_SUFFIX):
            data = _facts_from_file(path, document, schema)
            handoff_id = data["handoff"]["id"]
            canonical = _canonical_document(data)
        else:
            _validate_document(path, document)
            handoff_id = document["handoff"]["id"]
            canonical = _canonical_document(document)
            try:
                extracted = engine.extract(document, model)
            except engine.ExtractError as exc:
                raise InputError(f"model extraction failed ({exc})", str(path)) from exc
            data = {"handoff": {"id": handoff_id, "filename": document["handoff"]["filename"]}, **extracted}
        if handoff_id in entries:
            if entries[handoff_id][2] != canonical:
                raise InputError("conflicting documents share a handoff ID", str(path))
            continue  # identical repeated document collapses to one
        entries[handoff_id] = (path, data, canonical)

    identities: dict[str, tuple[str, str | None, Any]] = {}
    nodes: dict[str, Node] = {}
    enriched: dict[str, dict[str, Any]] = {}
    edges: set[Edge] = set()
    handoff_records: dict[str, tuple[str, ...]] = {}
    handoff_metadata: dict[str, tuple[str, ...]] = {}
    external: dict[str, str] = {}
    parents: dict[str, dict[str, str]] = {rel: {} for rel in schema.single_parent}
    fan_out: dict[tuple[str, str], set[str]] = {}
    edge_props: dict[tuple[str, str, str], tuple[tuple[str, Any], ...]] = {}
    typed_external: dict[str, tuple[str, Path]] = {}
    record_edges: list[tuple[str, tuple[str, ...], tuple[str, ...]]] = []
    for handoff_id in sorted(entries):
        path, data, _ = entries[handoff_id]
        from_facts = path.name.endswith(FACTS_SUFFIX)
        record_rels = tuple(schema.record_out_rels if from_facts else schema.deterministic_record_out_rels)
        containers = {e["end"]: e["start"] for e in data["edges"] if e["type"] in schema.containment}
        record_ids: list[str] = []
        for fact in data["nodes"]:
            ident, kind = fact["id"], fact["type"]
            parent = containers.get(ident)
            if model.nodes[kind].enrich:
                _register(identities, ident, kind, parent, None, path)
                merged = enriched.setdefault(ident, {})
                for key, value in fact["props"].items():
                    if merged.setdefault(key, value) != value:
                        raise InputError(f"{kind} identity maps to conflicting {key} values", str(path))
                nodes[ident] = Node(ident, kind, parent, tuple(sorted(merged.items())))
                continue
            frozen = tuple(sorted(fact["props"].items()))
            _register(identities, ident, kind, parent, frozen, path)
            nodes[ident] = Node(ident, kind, parent, frozen)
            if kind in schema.record_types and parent == handoff_id:
                record_ids.append(ident)
        handoff_records[handoff_id] = tuple(sorted(record_ids))
        references = {item["id"]: item["filename"] for item in data["external"] if item.get("type", schema.handoff_type) == schema.handoff_type}
        for item in data["external"]:
            if item.get("type", schema.handoff_type) != schema.handoff_type:
                typed_external.setdefault(item["id"], (item["type"], path))
        keep: list[str] = []
        record_keep: dict[str, list[str]] = {ident: [] for ident in record_ids}
        for fact in data["edges"]:
            props = tuple(sorted(fact.get("props", {}).items()))
            edge = Edge(fact["type"], fact["start"], fact["end"], props)
            if edge_props.setdefault((edge.type, edge.start, edge.end), props) != props:
                raise InputError(f"{edge.type} edge is supplied with conflicting properties", str(path))
            if edge.start in record_keep and edge.type in record_rels:
                record_keep[edge.start].append(f"{edge.type}|{edge.end}")
            if edge.end in references:
                filename = references[edge.end]
                if edge.end in entries:
                    if entries[edge.end][1]["handoff"]["filename"] != filename:
                        raise InputError("continuation endpoint has a conflicting filename", str(path))
                else:
                    external[edge.end] = filename
            if edge.type in parents and parents[edge.type].setdefault(edge.start, edge.end) != edge.end:
                raise InputError(f"{edge.type} source has conflicting parent targets", str(path))
            limit = schema.max_out.get(edge.type)
            if limit is not None:
                targets = fan_out.setdefault((edge.type, edge.start), set())
                targets.add(edge.end)
                if len(targets) > limit:
                    raise InputError(f"{edge.type} exceeds max_out {limit}", str(path))
            if edge.type in schema.metadata_rels and edge.start == handoff_id:
                keep.append(f"{edge.type}|{edge.end}")
            edges.add(edge)
        handoff = nodes.get(handoff_id)
        if handoff is None or handoff.kind != schema.handoff_type:
            raise InputError("input does not produce its handoff node", str(path))
        if handoff.props_dict().get("schema_version", 1) >= 2:
            handoff_metadata[handoff_id] = tuple(sorted(keep))
        if record_rels:
            record_edges.extend((ident, record_rels, tuple(sorted(record_keep[ident]))) for ident in sorted(record_keep))

    external_nodes: dict[str, str] = {}
    for ident, (kind, path) in sorted(typed_external.items()):
        if ident in nodes:
            if nodes[ident].kind != kind:
                raise InputError("an edge endpoint is supplied with a conflicting type", str(path))
        else:
            external_nodes[ident] = kind

    for rel, pairs in parents.items():
        _check_single_parent(pairs, {}, rel)
    child_of = tuple(sorted(parents.get("CHILD_OF", {}).items()))
    return Payload(
        sources=tuple(str(entries[h][0]) for h in sorted(entries)),
        nodes=tuple(nodes[i] for i in sorted(nodes)),
        edges=tuple(sorted(edges, key=lambda e: (e.type, e.start, e.end))),
        handoff_records=tuple(sorted(handoff_records.items())),
        handoff_metadata=tuple(sorted(handoff_metadata.items())),
        external_handoffs=tuple(sorted(external.items())),
        child_of=child_of,
        single_parent=tuple((rel, tuple(sorted(pairs.items()))) for rel, pairs in sorted(parents.items())),
        external_nodes=tuple(sorted(external_nodes.items())),
        record_edges=tuple(record_edges),
        schema=schema,
    )


# --------------------------------------------------------------------------- Cypher (constants; all values are parameters)
# Type-specific text (labels, relationship types) exists only in Schema.upsert and
# Schema.merge_edge; every other query receives model-derived sets as parameters.

Q_CONSTRAINT_IDENTITY = (
    "CREATE CONSTRAINT handoff_memory_identity IF NOT EXISTS "
    "FOR (n:HandoffMemory) REQUIRE (n.namespace, n.id) IS UNIQUE"
)
Q_CONSTRAINT_CONTROL = (
    "CREATE CONSTRAINT handoff_memory_control IF NOT EXISTS "
    "FOR (c:HandoffMemoryControl) REQUIRE c.namespace IS UNIQUE"
)
Q_LOCK = (
    "MERGE (c:HandoffMemoryControl {namespace: $ns}) "
    "SET c.writes = coalesce(c.writes, 0) + 1 "
    "RETURN c.writes AS writes"
)
Q_EXISTING = (
    "UNWIND $ids AS ident "
    "MATCH (n:HandoffMemory {namespace: $ns, id: ident}) "
    "OPTIONAL MATCH (p:HandoffMemory {namespace: $ns})-[r]->(n) "
    "WHERE type(r) IN $owned AND r.namespace = $ns "
    "RETURN n.id AS id, labels(n) AS labels, collect(p.id) AS parents"
)
Q_OBSOLETE = (
    "UNWIND $handoffs AS h "
    "MATCH (d:HandoffMemory {namespace: $ns, id: h.id})-[x]->(r:HandoffMemory {namespace: $ns}) "
    "WHERE type(x) IN $record_rels AND x.namespace = $ns "
    "AND all(l IN $handoff_labels WHERE l IN labels(d)) AND NOT r.id IN h.keep "
    "RETURN DISTINCT r.id AS id ORDER BY id"
)
_FOREIGN_WHERE = (
    "WHERE NOT (type(x) IN $owned AND coalesce(x.namespace, '') = $ns "
    "AND o:HandoffMemory AND coalesce(o.namespace, '') = $ns) "
)
Q_FOREIGN_INCIDENT = (
    "UNWIND $ids AS ident "
    "MATCH (n:HandoffMemory {namespace: $ns, id: ident})-[x]-(o) "
    + _FOREIGN_WHERE
    + "RETURN count(x) AS foreign"
)
Q_NAMESPACE_FOREIGN = (
    "MATCH (n:HandoffMemory {namespace: $ns})-[x]-(o) "
    + _FOREIGN_WHERE
    + "RETURN count(x) AS foreign"
)
Q_RECORD_TARGETS = (
    "UNWIND $ids AS ident "
    "MATCH (r:HandoffMemory {namespace: $ns, id: ident})-[x]->(t:HandoffMemory {namespace: $ns}) "
    "WHERE type(x) IN $rels AND x.namespace = $ns "
    "RETURN DISTINCT t.id AS id ORDER BY id"
)
Q_DELETE_NODES = (
    "UNWIND $ids AS ident "
    "MATCH (n:HandoffMemory {namespace: $ns, id: ident}) "
    "DETACH DELETE n"
)
Q_CLEAR_NAMESPACE = "MATCH (n:HandoffMemory {namespace: $ns}) DETACH DELETE n"
Q_STORED_HANDOFFS = (
    "UNWIND $ids AS ident "
    "OPTIONAL MATCH (n:HandoffMemory {namespace: $ns, id: ident}) "
    "RETURN ident AS id, "
    "CASE WHEN n IS NOT NULL AND all(l IN $handoff_labels WHERE l IN labels(n)) THEN n.filename END AS filename"
)
Q_SINGLE_PARENT_ALL = (
    "MATCH (c:HandoffMemory {namespace: $ns})-[r]->(p:HandoffMemory {namespace: $ns}) "
    "WHERE type(r) = $rel AND r.namespace = $ns "
    "RETURN c.id AS child, p.id AS parent"
)
Q_CHILD_OF_ALL = Q_SINGLE_PARENT_ALL
Q_STORED_NODES = (
    "UNWIND $ids AS ident "
    "OPTIONAL MATCH (n:HandoffMemory {namespace: $ns, id: ident}) "
    "RETURN ident AS id, labels(n) AS labels"
)
Q_STALE_RECORD_EDGES = (
    "UNWIND $records AS rec "
    "MATCH (x:HandoffMemory {namespace: $ns, id: rec.id})-[e]->(t:HandoffMemory {namespace: $ns}) "
    "WHERE type(e) IN rec.rels AND e.namespace = $ns AND NOT (type(e) + '|' + t.id) IN rec.keep "
    "DELETE e "
    "RETURN DISTINCT t.id AS id"
)
Q_STALE_METADATA = (
    "UNWIND $handoffs AS h "
    "MATCH (d:HandoffMemory {namespace: $ns, id: h.id})-[r]->(t:HandoffMemory {namespace: $ns}) "
    "WHERE type(r) IN $meta AND r.namespace = $ns AND all(l IN $handoff_labels WHERE l IN labels(d)) "
    "AND NOT (type(r) + '|' + t.id) IN h.keep "
    "DELETE r "
    "RETURN DISTINCT t.id AS id"
)
Q_DELETE_ORPHAN_CONTEXT = (
    "UNWIND $ids AS ident "
    "MATCH (n:HandoffMemory {namespace: $ns, id: ident}) "
    "WHERE any(l IN labels(n) WHERE l IN $keyed) AND NOT (n)--() "
    "DELETE n"
)


# --------------------------------------------------------------------------- transaction bodies

def _run(tx: Any, query: str, params: dict[str, Any]) -> list[dict[str, Any]]:
    return [record.data() for record in tx.run(query, params)]


def _acquire_writer_lock(tx: Any, schema: Schema) -> None:
    _run(tx, Q_LOCK, {"ns": schema.namespace})


def _check_conflicts(tx: Any, payload: Payload) -> None:
    schema = payload.active_schema
    expected = {node.id: node for node in payload.nodes}
    rows = _run(tx, Q_EXISTING, {"ns": schema.namespace, "ids": sorted(expected), "owned": schema.containment})
    for row in rows:
        node = expected[row["id"]]
        if sorted(row["labels"]) != sorted(schema.kind_labels[node.kind]):
            raise ConflictError("stored identity has a conflicting entity kind")
        parents = sorted(p for p in row["parents"] if p is not None)
        wanted = [] if node.parent is None else [node.parent]
        if parents != wanted:
            raise ConflictError("stored identity has conflicting parentage")


def _write_payload(tx: Any, payload: Payload) -> None:
    schema = payload.active_schema
    for kind in schema.node_order:  # model file order: containers before records
        rows = [{"id": n.id, "props": n.props_dict()} for n in payload.nodes_of(kind)]
        if rows:
            _run(tx, schema.upsert[kind], {"ns": schema.namespace, "rows": rows})
    for rel in schema.edge_order:
        rows = [{"start": e.start, "end": e.end, "props": dict(e.props)} for e in payload.edges_of(rel)]
        if rows:
            _run(tx, schema.merge_edge[rel], {"ns": schema.namespace, "rows": rows})


def _check_stored_endpoints(tx: Any, payload: Payload) -> None:
    schema = payload.active_schema
    if payload.external_handoffs:
        wanted = dict(payload.external_handoffs)
        rows = _run(tx, Q_STORED_HANDOFFS, {"ns": schema.namespace, "ids": sorted(wanted), "handoff_labels": schema.handoff_labels})
        found = {row["id"]: row["filename"] for row in rows if row["filename"] is not None}
        for ident, filename in wanted.items():
            if found.get(ident) != filename:
                raise ConflictError("a continuation references a handoff that is neither in the input nor stored")
    if payload.external_nodes:
        wanted_nodes = dict(payload.external_nodes)
        rows = _run(tx, Q_STORED_NODES, {"ns": schema.namespace, "ids": sorted(wanted_nodes)})
        found_labels = {row["id"]: sorted(row["labels"] or []) for row in rows}
        for ident, kind in wanted_nodes.items():
            if found_labels.get(ident) != sorted(schema.kind_labels[kind]):
                raise ConflictError("a relationship references a node that is neither in the input nor stored")
    batch = dict(payload.single_parent)
    for rel in schema.single_parent:
        stored = {row["child"]: row["parent"] for row in _run(tx, Q_SINGLE_PARENT_ALL, {"ns": schema.namespace, "rel": rel})}
        try:
            _check_single_parent(dict(batch.get(rel, ())), stored, rel)
        except InputError as exc:
            raise ConflictError(str(exc)) from exc


def _reconcile_metadata(tx: Any, payload: Payload) -> list[str]:
    schema = payload.active_schema
    handoffs = [{"id": hid, "keep": list(keep)} for hid, keep in payload.handoff_metadata]
    if not handoffs:
        return []
    params = {"ns": schema.namespace, "handoffs": handoffs, "meta": schema.metadata_rels, "handoff_labels": schema.handoff_labels}
    return [row["id"] for row in _run(tx, Q_STALE_METADATA, params)]


def import_tx(tx: Any, payload: Payload) -> dict[str, int]:
    """Managed write callback: idempotent, reads only the frozen payload."""
    schema = payload.active_schema
    ns = schema.namespace
    _acquire_writer_lock(tx, schema)
    _check_conflicts(tx, payload)
    _check_stored_endpoints(tx, payload)
    handoffs = [{"id": hid, "keep": list(keep)} for hid, keep in payload.handoff_records]
    obsolete_params = {"ns": ns, "handoffs": handoffs, "record_rels": schema.record_rels, "handoff_labels": schema.handoff_labels}
    obsolete = [row["id"] for row in _run(tx, Q_OBSOLETE, obsolete_params)]
    orphans: list[str] = []
    if obsolete:
        foreign = _run(tx, Q_FOREIGN_INCIDENT, {"ns": ns, "ids": obsolete, "owned": schema.owned})
        if foreign and foreign[0]["foreign"]:
            raise BoundaryError("an obsolete record has a foreign relationship; nothing was changed")
        if schema.record_out_rels:
            orphans += [row["id"] for row in _run(tx, Q_RECORD_TARGETS, {"ns": ns, "ids": obsolete, "rels": schema.record_out_rels})]
        _run(tx, Q_DELETE_NODES, {"ns": ns, "ids": obsolete})
    orphans += _reconcile_metadata(tx, payload)
    if payload.record_edges:
        records = [{"id": ident, "rels": list(rels), "keep": list(keep)} for ident, rels, keep in payload.record_edges]
        orphans += [row["id"] for row in _run(tx, Q_STALE_RECORD_EDGES, {"ns": ns, "records": records})]
    _write_payload(tx, payload)
    if orphans:
        _run(tx, Q_DELETE_ORPHAN_CONTEXT, {"ns": ns, "ids": sorted(set(orphans)), "keyed": schema.key_labels})
    return {"removed": len(obsolete)}


def rebuild_tx(tx: Any, payload: Payload) -> dict[str, int]:
    """Managed write callback: full namespace replacement inside one transaction."""
    schema = payload.active_schema
    _acquire_writer_lock(tx, schema)
    foreign = _run(tx, Q_NAMESPACE_FOREIGN, {"ns": schema.namespace, "owned": schema.owned})
    if foreign and foreign[0]["foreign"]:
        raise BoundaryError("a relationship crosses the handoff namespace boundary; nothing was deleted")
    if payload.external_handoffs:
        raise ConflictError("rebuild input references a continuation outside the supplied set")
    if payload.external_nodes:
        raise ConflictError("rebuild input references a node outside the supplied set")
    _run(tx, Q_CLEAR_NAMESPACE, {"ns": schema.namespace})
    _write_payload(tx, payload)
    return {"removed": 0}


def ensure_schema(session: Any) -> None:
    for query in (Q_CONSTRAINT_IDENTITY, Q_CONSTRAINT_CONTROL):
        session.execute_write(lambda tx, q=query: _run(tx, q, {}))


# --------------------------------------------------------------------------- connection

@dataclass(frozen=True)
class Connection:
    uri: str
    user: str
    database: str
    password: str

    def __repr__(self) -> str:  # never expose the password
        return f"Connection(uri={self.uri!r}, user={self.user!r}, database={self.database!r})"


def connection_from(args: argparse.Namespace, environ: dict[str, str] | None = None) -> Connection:
    environ = os.environ if environ is None else environ
    parts = urlsplit(args.uri)
    if parts.scheme not in ALLOWED_SCHEMES or (parts.hostname or "") not in LOOPBACK_HOSTS:
        raise ConfigError("only loopback bolt:// or neo4j:// endpoints are allowed")
    if parts.username or parts.password:
        raise ConfigError("credentials must not be embedded in the URI")
    password = environ.get("NEO4J_PASSWORD", "")
    if not password:
        raise ConfigError("NEO4J_PASSWORD is not set")
    return Connection(args.uri, args.user, args.database, password)


def default_driver_factory(conn: Connection, bounded: bool = False) -> Any:
    try:
        from neo4j import GraphDatabase
    except ImportError as exc:
        raise ConfigError("the neo4j driver is not installed (see runtime/requirements.txt of the handoff skill)") from exc
    # absent-label/type warnings are expected before the first v2 import; keep CLI output to the result line
    timeouts = {"connection_timeout": 5, "connection_acquisition_timeout": 10,
                "max_transaction_retry_time": 5} if bounded else {}
    return GraphDatabase.driver(conn.uri, auth=(conn.user, conn.password),
                                notifications_min_severity="OFF", **timeouts)


def apply(mode: str, payload: Payload, conn: Connection, driver_factory: Callable[[Connection], Any]) -> dict[str, int]:
    callback = import_tx if mode == "import" else rebuild_tx
    driver = driver_factory(conn)
    try:
        with driver.session(database=conn.database) as session:
            ensure_schema(session)
            return session.execute_write(callback, payload)
    finally:
        driver.close()


# --------------------------------------------------------------------------- CLI

def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Validate/import handoff graph sidecars (v1, v2) and facts into local Neo4j through the graph model")
    sub = parser.add_subparsers(dest="command", required=True)
    for name, help_text in (
        ("validate", "offline validation only"),
        ("import", "import and reconcile the supplied handoffs"),
        ("rebuild", "replace the whole handoff namespace with the supplied set"),
    ):
        cmd = sub.add_parser(name, help=help_text)
        cmd.add_argument("paths", nargs="+", help="sidecar/facts files or directories of HANDOFF-*.graph.json")
        cmd.add_argument("--facts", action="store_true", help="directories yield HANDOFF-*.facts.json instead of sidecars")
        cmd.add_argument("--model", default=None, help="graph model file (default: references/graph-model.yaml of the handoff skill)")
        if name != "validate":
            cmd.add_argument("--uri", default=DEFAULT_URI)
            cmd.add_argument("--user", default=DEFAULT_USER)
            cmd.add_argument("--database", default=DEFAULT_DATABASE)
    for name in ("import-saved", "import-pending"):
        cmd = sub.add_parser(name, help="import saved handoff or retry the private queue")
        if name == "import-saved":
            cmd.add_argument("handoff", help="saved HANDOFF-NNN.md")
        cmd.add_argument("--model", default=None)
        cmd.add_argument("--uri", default=DEFAULT_URI)
        cmd.add_argument("--user", default=DEFAULT_USER)
        cmd.add_argument("--database", default=DEFAULT_DATABASE)
    return parser


def _safe_db_error(exc: BaseException, secret: str = "") -> str:
    # Only known categories, never user-controlled messages/class names/codes.
    name = type(exc).__name__
    safe = name if name in {"ServiceUnavailable", "AuthError", "SessionExpired",
                           "TransientError", "ClientError", "DatabaseError"} else "DatabaseError"
    code = getattr(exc, "code", None)
    if (isinstance(code, str) and len(code) <= 100 and
            re.fullmatch(r"Neo\.(ClientError|TransientError|DatabaseError)\.[A-Za-z]+\.[A-Za-z]+", code)
            and not (secret and secret in code)):
        safe += " " + code
    return safe


class QueueError(ImportFailure):
    """Private state is unsafe, malformed or cannot be persisted."""


def queue_file(environ: dict[str, str]) -> Path:
    state = environ.get("XDG_STATE_HOME") or str(Path(environ.get("HOME", str(Path.home()))) / ".local/state")
    return Path(environ.get("HANDOFF_MEMORY_QUEUE_FILE") or
                str(Path(state) / "handoff-memory/pending-imports.json")).expanduser().resolve()


def _safe_queue_location(path: Path) -> None:
    # Check resolved location, including all ephemeral files, before mkdir/write.
    ancestor = path.parent
    while not ancestor.exists() and ancestor != ancestor.parent:
        ancestor = ancestor.parent
    result = subprocess.run(["git", "-C", str(ancestor), "rev-parse", "--show-toplevel"],
                            capture_output=True, text=True)
    if result.returncode != 0:
        return
    root = Path(result.stdout.strip()).resolve()
    candidates = [path, path.with_name(path.name + ".lock"),
                  path.with_name(path.name + ".tmp-probe")]
    # Require the directory itself to be ignored: every random replacement is safe.
    candidates.append(path.parent / "pending-replacement-probe")
    for candidate in candidates:
        checked = subprocess.run(["git", "-C", str(root), "check-ignore", "--quiet", str(candidate)],
                                 capture_output=True)
        if checked.returncode != 0:
            raise QueueError("unsafe queue location")


class PendingQueue:
    def __init__(self, path: Path) -> None:
        self.path = path
        self.paths: set[str] = set()

    def read(self) -> None:
        if not self.path.exists():
            self.paths = set()
            return
        data = json.loads(self.path.read_text(encoding="utf-8"), object_pairs_hook=_no_duplicate_keys,
                          parse_constant=_reject_constant)
        if (not isinstance(data, dict) or set(data) != {"version", "paths"} or
                type(data["version"]) is not int or data["version"] != 1 or
                not isinstance(data["paths"], list)):
            raise QueueError("invalid queue")
        entries = data["paths"]
        if any(not isinstance(p, str) or not Path(p).is_absolute() or
               not p.endswith(SIDECAR_SUFFIX) or str(Path(p).resolve()) != p for p in entries):
            raise QueueError("invalid queue")
        self.paths = set(entries)

    def write(self, paths: set[str]) -> None:
        tmp = None
        backup = None
        replaced = False
        retain_backup = False
        directory_fd = os.open(self.path.parent, os.O_RDONLY | os.O_DIRECTORY)
        try:
            # Reject unsupported directory fsync before changing the queue.
            os.fsync(directory_fd)
            if self.path.exists():
                backup = self.path.with_name(self.path.name + ".tmp-prior-" + uuid.uuid4().hex)
                os.link(self.path, backup)
                os.fsync(directory_fd)  # prior state remains recoverable even if restoration fails
            fd, name = tempfile.mkstemp(prefix=self.path.name + ".tmp-", dir=self.path.parent)
            tmp = Path(name)
            with os.fdopen(fd, "w", encoding="utf-8") as stream:
                json.dump({"version": 1, "paths": sorted(paths)}, stream)
                stream.write("\n")
                stream.flush()
                os.fsync(stream.fileno())
            os.replace(tmp, self.path)
            replaced = True
            os.fsync(directory_fd)  # persist the new directory entry, not just file contents
            self.paths = set(paths)
        except OSError as exc:
            if replaced:
                try:
                    if backup is not None:
                        os.replace(backup, self.path)
                        backup = None
                    else:
                        self.path.unlink()
                    try:
                        os.fsync(directory_fd)
                    except OSError:
                        pass  # restored bytes remain; storage failure is still reported below
                except OSError:
                    retain_backup = backup is not None
            raise QueueError("queue storage failure") from exc
        finally:
            os.close(directory_fd)
            if tmp is not None:
                tmp.unlink(missing_ok=True)
            if backup is not None and not retain_backup:
                backup.unlink(missing_ok=True)

    @contextlib.contextmanager
    def locked(self):
        try:
            _safe_queue_location(self.path)
            self.path.parent.mkdir(parents=True, mode=0o700, exist_ok=True)
            if self.path.parent.stat().st_mode & 0o077:
                raise QueueError("queue requires a private directory")
            lock = self.path.with_name(self.path.name + ".lock")
            # Never follow an attacker-created lock-file symlink.
            fd = os.open(lock, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
            with os.fdopen(fd, "a") as stream:
                os.fchmod(stream.fileno(), 0o600)
                fcntl.flock(stream, fcntl.LOCK_EX)
                self.read()
                if self.path.exists():
                    self.path.chmod(0o600)
                yield self
        except QueueError:
            raise
        except (OSError, ValueError, TypeError) as exc:
            raise QueueError("queue storage failure") from exc


def saved_selection(root: Path, schema: Schema) -> tuple[Payload, set[str]]:
    selected: set[Path] = set()
    documents: dict[Path, Any] = {}
    stack = [root.resolve()]
    while stack:
        path = stack.pop()
        if path in selected:
            continue
        selected.add(path)
        doc = load_strict(path)[1]
        _validate_document(path, doc)
        documents[path] = doc
        if doc.get("schema_version") == 2:
            predecessor = doc.get("metadata", {}).get("continues")
            if predecessor is not None:
                if not isinstance(predecessor, str) or not re.fullmatch(r"HANDOFF-[0-9]+", predecessor):
                    raise InputError("invalid continuation")
                parent = path.parent / (predecessor + SIDECAR_SUFFIX)
                if parent.is_file():
                    stack.append(parent.resolve())
    return build_payload(sorted(selected), schema, frozen=documents), {str(p) for p in selected}


def automatic_connection(args: argparse.Namespace, environ: dict[str, str]) -> Connection:
    values = dict(environ)
    if not values.get("NEO4J_PASSWORD"):
        try:
            home = Path(values.get("HOME", str(Path.home())))
            values["NEO4J_PASSWORD"] = (home / ".config/handoff-memory/neo4j-password").read_text().rstrip("\r\n")
        except (OSError, UnicodeError):
            pass
    return connection_from(args, values)


def automatic_import(args: argparse.Namespace, schema: Schema,
                     factory: Callable[[Connection], Any], environ: dict[str, str]) -> int:
    current = None
    if args.command == "import-saved":
        try:
            md = Path(args.handoff).resolve()
            if not re.fullmatch(r"HANDOFF-[0-9]+\.md", md.name) or not md.is_file():
                raise InputError("invalid saved handoff")
            current = str(md.with_name(md.stem + SIDECAR_SUFFIX).resolve())
            build_payload([current], schema)
        except Exception:
            print("validation failed: invalid saved selection", file=sys.stderr)
            return EXIT_INPUT
    try:
        queue = PendingQueue(queue_file(environ))
        with queue.locked():
            if current:
                queue.write(queue.paths | {current})
            order = ([current] if current else []) + sorted(queue.paths - ({current} if current else set()))
            committed_paths: set[str] = set()
            for root in order:
                if root not in queue.paths:
                    continue
                try:
                    payload, supplied = saved_selection(Path(root), schema)
                except Exception:
                    print("pending import: invalid input", file=sys.stderr)
                    continue
                conn = None
                try:
                    conn = automatic_connection(args, environ)
                    apply("import", payload, conn, factory)
                except ConfigError:
                    print("pending import: configuration unavailable", file=sys.stderr)
                    continue
                except Exception as exc:
                    print("pending import: database operation error (" +
                          _safe_db_error(exc, conn.password if conn else "") + "); no selection committed", file=sys.stderr)
                    continue
                committed_paths.update(supplied)
                try:
                    queue.write(queue.paths - supplied)
                except (OSError, QueueError):
                    print("database committed; pending acknowledgement storage failed", file=sys.stderr)
                    if current:
                        print(f"{Path(args.handoff).name}: {'imported' if current in committed_paths else 'da importare'}; pending {len(queue.paths)}; acknowledgement deferred")
                        return EXIT_OK
                    return EXIT_DATABASE
            count = len(queue.paths)
            if current:
                print(f"{Path(args.handoff).name}: {'da importare' if current in queue.paths else 'imported'}; pending {count}")
                return EXIT_OK
            print(f"import-pending: pending {count}")
            return EXIT_DATABASE if count else EXIT_OK
    except Exception:
        print("queue storage failure: pending state unavailable", file=sys.stderr)
        return EXIT_DATABASE


@contextlib.contextmanager
def explicit_acknowledgement(args: argparse.Namespace, environ: dict[str, str]):
    """Legacy import remains usable when private queue state is unavailable."""
    queue = None
    manager = None
    if args.command == "import" and not args.facts:
        try:
            candidate = PendingQueue(queue_file(environ))
            if candidate.path.exists():
                manager = candidate.locked()
                queue = manager.__enter__()
        except Exception:
            manager = None
            print("warning: pending acknowledgement unavailable", file=sys.stderr)
    try:
        yield queue
    finally:
        if manager is not None:
            manager.__exit__(None, None, None)


def main(argv: list[str] | None = None, driver_factory: Callable[[Connection], Any] | None = None,
         environ: dict[str, str] | None = None) -> int:
    args = _parser().parse_args(argv)
    try:
        schema = SCHEMA if args.model is None else load_schema(args.model)
    except engine.ModelError as exc:
        print(f"validation failed: {exc}", file=sys.stderr)
        return EXIT_INPUT
    if args.command in {"import-saved", "import-pending"}:
        return automatic_import(args, schema, driver_factory or (lambda conn: default_driver_factory(conn, bounded=True)),
                                dict(os.environ if environ is None else environ))
    with explicit_acknowledgement(args, dict(os.environ if environ is None else environ)) as queue:
        try:
            supplied_paths = discover(args.paths, facts=args.facts)
            payload = build_payload(supplied_paths, schema, facts=args.facts)
        except InputError as exc:
            print(f"validation failed: {exc}", file=sys.stderr)
            return EXIT_INPUT
        handoffs = len(payload.handoff_records)
        records = sum(len(keep) for _, keep in payload.handoff_records)
        if args.command == "validate":
            print(f"valid: {handoffs} handoff(s), {records} record(s), payload {payload.digest()}")
            return EXIT_OK
        try:
            conn = connection_from(args, environ)
            result = apply(args.command, payload, conn, driver_factory or default_driver_factory)
            if queue is not None:
                try:
                    queue.write(queue.paths - {str(p) for p in supplied_paths})
                except Exception:
                    print("warning: database committed; pending acknowledgement storage failed", file=sys.stderr)
        except (ConfigError, ConflictError, BoundaryError) as exc:
            print(f"{args.command} failed: {exc}", file=sys.stderr)
            return EXIT_DATABASE
        except Exception as exc:  # noqa: BLE001 - report only the error class/code
            print(f"{args.command} failed: database operation error ({_safe_db_error(exc, conn.password if 'conn' in locals() else '')}); no changes committed", file=sys.stderr)
            return EXIT_DATABASE
        print(
            f"{args.command}: {handoffs} handoff(s), {records} record(s) into namespace {schema.namespace}; "
            f"removed {result.get('removed', 0)} obsolete record(s); payload {payload.digest()}"
        )
        return EXIT_OK


if __name__ == "__main__":
    raise SystemExit(main())