#!/usr/bin/env python3
"""W4 automation helpers for agentctl: durable jobs, doc projection, deploy loop.

Linux-first. Process liveness reuses the host agentctl proc/alive/boot helpers
passed in as callbacks so a future macOS process layer can plug in without
rewriting queue semantics.
"""
from __future__ import annotations

import json
import os
import re
import sqlite3
import time
import uuid
from pathlib import Path
from typing import Any, Callable, Optional

SCHEMA_SQL = '''
CREATE TABLE IF NOT EXISTS jobs(
  id TEXT PRIMARY KEY,
  task TEXT NOT NULL,
  kind TEXT NOT NULL,
  payload TEXT NOT NULL,
  state TEXT NOT NULL,
  priority INTEGER NOT NULL DEFAULT 100,
  not_before REAL NOT NULL,
  created REAL NOT NULL,
  updated REAL NOT NULL,
  lease_run TEXT,
  attempts INTEGER NOT NULL DEFAULT 0,
  last_error TEXT,
  result TEXT
);
CREATE INDEX IF NOT EXISTS jobs_ready ON jobs(state, not_before, priority, created);
CREATE TABLE IF NOT EXISTS deployments(
  id TEXT PRIMARY KEY,
  task TEXT NOT NULL,
  env TEXT NOT NULL,
  expected_commit TEXT NOT NULL,
  manifest TEXT,
  phase TEXT NOT NULL,
  dry_run INTEGER NOT NULL DEFAULT 0,
  created REAL NOT NULL,
  updated REAL NOT NULL,
  receipt TEXT,
  last_error TEXT
);
CREATE INDEX IF NOT EXISTS deployments_task ON deployments(task, phase);
CREATE TABLE IF NOT EXISTS projections(
  file TEXT NOT NULL,
  record_id TEXT NOT NULL,
  task TEXT NOT NULL,
  source TEXT NOT NULL,
  content_hash TEXT NOT NULL,
  updated REAL NOT NULL,
  PRIMARY KEY(file, record_id)
);
'''

JOB_KINDS = frozenset({
    'validate', 'integrate', 'with', 'deploy_prepare', 'deploy_run',
    'deploy_accept', 'docs_project', 'cleanup_inspect', 'noop',
})
JOB_ACTIVE = ('pending', 'leased', 'running')
JOB_TERMINAL = ('completed', 'failed', 'cancelled')
DEPLOY_PHASES = (
    'prepared', 'deploying', 'accepting', 'passed', 'failed', 'recovery_required',
)
# Beyond CURRENT/MATRIX: any Markdown table under .doc/ may be projected when
# registered here. Layout changes still require docs:admin via docs apply.
PROJECTABLE_DOCS = frozenset({
    '.doc/ops/CURRENT.md',
    '.doc/status/FEATURE-MATRIX.md',
    '.doc/ops/CURRENT.md',  # duplicate ok in set
})
PROJECTABLE_PREFIXES = ('.doc/ops/', '.doc/status/', '.doc/remediation/')


def ensure_schema(db: sqlite3.Connection) -> None:
    db.executescript(SCHEMA_SQL)


def _now() -> float:
    return time.time()


def queue_enqueue(db, *, task: str, kind: str, payload: dict, priority: int = 100,
                  not_before: Optional[float] = None, job_id: Optional[str] = None) -> dict:
    if kind not in JOB_KINDS:
        raise ValueError(f'unsupported job kind: {kind}')
    if not isinstance(payload, dict):
        raise ValueError('payload must be an object')
    jid = job_id or str(uuid.uuid4())
    if not re.fullmatch(r'[A-Za-z0-9._-]{1,128}', jid):
        raise ValueError('invalid job id')
    ts = _now()
    when = float(not_before) if not_before is not None else ts
    existing = db.execute('SELECT * FROM jobs WHERE id=?', (jid,)).fetchone()
    if existing:
        if (existing['task'] != task or existing['kind'] != kind
                or existing['payload'] != json.dumps(payload, sort_keys=True)):
            raise ValueError('job id reused with different payload')
        return dict(existing)
    db.execute(
        '''INSERT INTO jobs(id,task,kind,payload,state,priority,not_before,created,updated)
           VALUES(?,?,?,?,?,?,?,?,?)''',
        (jid, task, kind, json.dumps(payload, sort_keys=True), 'pending',
         int(priority), when, ts, ts))
    return dict(db.execute('SELECT * FROM jobs WHERE id=?', (jid,)).fetchone())


def queue_list(db, *, task: Optional[str] = None, state: Optional[str] = None,
               limit: int = 50) -> list:
    clauses, args = [], []
    if task:
        clauses.append('task=?')
        args.append(task)
    if state:
        clauses.append('state=?')
        args.append(state)
    where = (' WHERE ' + ' AND '.join(clauses)) if clauses else ''
    rows = db.execute(
        f'SELECT * FROM jobs{where} ORDER BY priority ASC, created ASC LIMIT ?',
        (*args, max(1, min(int(limit), 500)))).fetchall()
    return [dict(r) for r in rows]


def queue_cancel(db, job_id: str, reason: str) -> dict:
    row = db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone()
    if not row:
        raise KeyError(job_id)
    if row['state'] in JOB_TERMINAL:
        return dict(row)
    if row['state'] == 'running':
        raise RuntimeError('cannot cancel a running job; recover the lease first')
    db.execute(
        "UPDATE jobs SET state='cancelled',last_error=?,updated=? WHERE id=?",
        (reason, _now(), job_id))
    return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone())


def claim_next_job(db, *, worker: str) -> Optional[dict]:
    """Atomically lease the next due pending job. Caller must be in a write txn."""
    ts = _now()
    row = db.execute(
        '''SELECT * FROM jobs WHERE state='pending' AND not_before<=?
           ORDER BY priority ASC, created ASC LIMIT 1''', (ts,)).fetchone()
    if not row:
        return None
    db.execute(
        '''UPDATE jobs SET state='leased',lease_run=?,attempts=attempts+1,updated=?
           WHERE id=? AND state='pending' ''',
        (worker, ts, row['id']))
    claimed = db.execute('SELECT * FROM jobs WHERE id=?', (row['id'],)).fetchone()
    if not claimed or claimed['state'] != 'leased':
        return None
    return dict(claimed)


def finish_job(db, job_id: str, *, ok: bool, result: Optional[dict] = None,
               error: Optional[str] = None) -> dict:
    state = 'completed' if ok else 'failed'
    db.execute(
        '''UPDATE jobs SET state=?,result=?,last_error=?,updated=? WHERE id=?''',
        (state, json.dumps(result or {}, sort_keys=True), error, _now(), job_id))
    return dict(db.execute('SELECT * FROM jobs WHERE id=?', (job_id,)).fetchone())


def mark_job_running(db, job_id: str) -> None:
    db.execute("UPDATE jobs SET state='running',updated=? WHERE id=?", (_now(), job_id))


def projectable_file(path: str) -> bool:
    if path in PROJECTABLE_DOCS or path in {'.doc/ops/CURRENT.md', '.doc/status/FEATURE-MATRIX.md'}:
        return True
    return any(path.startswith(prefix) for prefix in PROJECTABLE_PREFIXES) and path.endswith('.md')


def build_projection_row(*, task_id: str, owner: str, branch: str, worktree: str,
                         mutex: str, status: str, extra: Optional[dict] = None) -> str:
    """Canonical CURRENT-style row used for machine→doc projection."""
    cells = [owner, task_id, branch, worktree, mutex, status]
    if extra:
        for key in sorted(extra):
            cells.append(str(extra[key]))
    return '| ' + ' | '.join(cells) + ' |'


def record_projection(db, *, file: str, record_id: str, task: str, source: str,
                      content_hash: str) -> None:
    db.execute(
        '''INSERT INTO projections(file,record_id,task,source,content_hash,updated)
           VALUES(?,?,?,?,?,?)
           ON CONFLICT(file,record_id) DO UPDATE SET
             task=excluded.task, source=excluded.source,
             content_hash=excluded.content_hash, updated=excluded.updated''',
        (file, record_id, task, source, content_hash, _now()))


def deployment_create(db, *, deploy_id: str, task: str, env: str, expected_commit: str,
                      manifest: Optional[str], dry_run: bool) -> dict:
    if env not in ('dev', 'test', 'main'):
        raise ValueError('env must be dev|test|main')
    if not re.fullmatch(r'[0-9a-f]{7,40}', expected_commit):
        raise ValueError('expected_commit must be a hex SHA')
    ts = _now()
    existing = db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone()
    if existing:
        if (existing['task'] != task or existing['env'] != env
                or existing['expected_commit'] != expected_commit):
            raise ValueError('deploy id reused with different identity')
        return dict(existing)
    db.execute(
        '''INSERT INTO deployments(id,task,env,expected_commit,manifest,phase,dry_run,created,updated)
           VALUES(?,?,?,?,?,?,?,?,?)''',
        (deploy_id, task, env, expected_commit, manifest, 'prepared',
         1 if dry_run else 0, ts, ts))
    return dict(db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone())


def deployment_set_phase(db, deploy_id: str, phase: str, *, receipt: Optional[dict] = None,
                         error: Optional[str] = None) -> dict:
    if phase not in DEPLOY_PHASES:
        raise ValueError(f'invalid deploy phase: {phase}')
    db.execute(
        '''UPDATE deployments SET phase=?,receipt=?,last_error=?,updated=? WHERE id=?''',
        (phase, json.dumps(receipt or {}, sort_keys=True) if receipt is not None else None,
         error, _now(), deploy_id))
    return dict(db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone())


def deployment_get(db, deploy_id: str) -> Optional[dict]:
    row = db.execute('SELECT * FROM deployments WHERE id=?', (deploy_id,)).fetchone()
    return dict(row) if row else None


def dry_run_deploy_steps(deploy: dict) -> dict:
    """Closed-loop steps without touching shared docker. Used by tests and design."""
    return {
        'env': deploy['env'],
        'expected_commit': deploy['expected_commit'],
        'manifest': deploy.get('manifest'),
        'steps': [
            {'name': 'guard_env_lease', 'status': 'ok'},
            {'name': 'verify_main_clean', 'status': 'ok'},
            {'name': 'apply_or_rebuild', 'status': 'skipped_dry_run'},
            {'name': 'health_ready', 'status': 'skipped_dry_run'},
            {'name': 'pi_image_identity', 'status': 'skipped_dry_run'},
        ],
        'dry_run': True,
    }


def cleanup_visibility_report(
    *,
    path: Path,
    branch: str,
    head: str,
    ancestor_of_dev: bool,
    dirty: bool,
    unfinished_git: list,
    registry: dict,
    process: dict,
) -> dict:
    """Structured cleanup gate report. Never deletes; visibility only."""
    blockers = []
    if not ancestor_of_dev:
        blockers.append({'code': 'NOT_MERGED', 'message': 'HEAD is not an ancestor of dev'})
    if dirty:
        blockers.append({'code': 'DIRTY_TREE', 'message': 'tracked/untracked/ignored files remain'})
    for marker in unfinished_git:
        blockers.append({'code': 'GIT_IN_PROGRESS', 'message': f'unfinished Git operation: {marker}'})
    if registry.get('active_task'):
        blockers.append({'code': 'TASK_ACTIVE', 'message': 'registered task still active'})
    if registry.get('live_run'):
        blockers.append({'code': 'LOCK_BUSY', 'message': 'live or unresolved operation',
                         'run_id': registry.get('live_run')})
    status = process.get('status')
    if status == 'busy':
        blockers.append({'code': 'PROCESS_BUSY', 'message': 'process still uses worktree cwd',
                         'pids': process.get('pids', [])})
    elif status == 'unreadable':
        blockers.append({'code': 'PROCESS_UNREADABLE', 'message': 'cannot verify same-user cwd',
                         'pids': process.get('unreadable', []),
                         'platform_note': process.get('platform_note')})
    elif status == 'unsupported':
        blockers.append({'code': 'PROCESS_UNSUPPORTED',
                         'message': process.get('message', 'process inspection unsupported'),
                         'platform_note': process.get('platform_note')})
    return {
        'status': 'ready' if not blockers else 'cleanup_pending',
        'path': str(path),
        'branch': branch,
        'head': head,
        'gates': {
            'ancestor_of_dev': ancestor_of_dev,
            'dirty': dirty,
            'unfinished_git': unfinished_git,
            'registry': registry,
            'process': process,
        },
        'blockers': blockers,
    }


def linux_cwd_visibility(path: Path, uid: Optional[int] = None) -> dict:
    """Fail-closed cwd scan using /proc. macOS should use lsof via sibling worktree."""
    import sys
    if not sys.platform.startswith('linux'):
        return {
            'status': 'unsupported',
            'message': f'/proc cwd scan unavailable on {sys.platform}',
            'platform_note': 'Use agentctl-macos lsof enumerator; treat unknown as cleanup_pending',
            'pids': [],
            'unreadable': [],
        }
    uid = os.getuid() if uid is None else uid
    busy, unreadable = [], []
    try:
        entries = list(Path('/proc').iterdir())
    except OSError:
        return {
            'status': 'unreadable',
            'message': 'cannot list /proc',
            'platform_note': 'fail-closed',
            'pids': [],
            'unreadable': ['*'],
        }
    for proc in entries:
        if not proc.name.isdigit():
            continue
        try:
            if proc.stat().st_uid != uid:
                continue
            cwd = Path(os.readlink(proc / 'cwd'))
            if cwd == path or path in cwd.parents:
                busy.append(proc.name)
        except (FileNotFoundError, ProcessLookupError):
            continue
        except PermissionError:
            try:
                if (proc / 'comm').read_text().strip() == 'systemd':
                    continue
            except (OSError, UnicodeError):
                pass
            unreadable.append(proc.name)
    if busy:
        return {'status': 'busy', 'pids': busy, 'unreadable': unreadable}
    if unreadable:
        return {'status': 'unreadable', 'pids': [], 'unreadable': unreadable,
                'platform_note': 'fail-closed; do not treat as exited'}
    return {'status': 'clear', 'pids': [], 'unreadable': []}


def pi_source_bind_paths() -> tuple:
    """Host paths that must not be bind-mounted when PI image delivery is enforced."""
    return (
        './deploy/pi-agent/runner.mjs',
        './deploy/pi-agent/pi-args.mjs',
        './deploy/pi-agent/result-v1.mjs',
        './deploy/pi-agent/extensions',
        'deploy/pi-agent/runner.mjs',
        'deploy/pi-agent/pi-args.mjs',
        'deploy/pi-agent/result-v1.mjs',
        'deploy/pi-agent/extensions',
    )


def inspect_compose_for_pi_binds(compose_text: str) -> dict:
    """Static check: flag source bind mounts under pi-agent service."""
    forbidden = pi_source_bind_paths()
    hits = []
    in_pi = False
    for line in compose_text.splitlines():
        stripped = line.strip()
        if re.match(r'^pi-agent\s*:', stripped):
            in_pi = True
            continue
        if in_pi and re.match(r'^[A-Za-z0-9_.-]+:\s*$', stripped):
            in_pi = False
        if in_pi:
            for item in forbidden:
                if item in stripped and (':/app/' in stripped or stripped.startswith('-')):
                    hits.append(stripped)
                    break
    return {
        'ok': not hits,
        'source_binds': hits,
        'recommendation': (
            'Apply docker-compose.dev.pi-image.yml overlay so runner source ships in the image'
            if hits else 'no pi-agent source binds detected'
        ),
    }


# Type aliases for agentctl callbacks (avoid circular imports at runtime).
EmitFn = Callable[[Any], None]
DenyFn = Callable[..., None]