#!/usr/bin/env python3
"""velcro-recordings: turn every recording `velcro import` put in a folder into text, then hand it on.

`velcro recordings run` runs it with your settings; it also runs on its own:

  velcro-recordings run    --dir <folder> --out <folder> --state <folder> [options]
  velcro-recordings status --dir <folder> --out <folder> --state <folder> [options]

Each line of <dir>/.velcro-import.jsonl is one recording. For each one, in order:

  1. transcribe: check the file against the SHA-256 the import wrote, then run the engine (no default).
     --engine claude or codex runs `velcro transcribe --engine` with it (your own login); anything else is a command
     line run with the file: {file} and {json} in it are replaced by the recording and a JSON file
     the command may write, otherwise the file goes last. The command prints the text.
     Writes <out>/<name>.txt, then <out>/<name>.json (written last: it marks the step done).
  2. tidy (optional): --tidy <command>, run with the VELCRO_* variables below. What it prints is
     kept in <out>/<name>.tidy.txt.
  3. deliver (optional): --deliver <command>, the same, also with VELCRO_TIDIED. What it prints is
     kept in <out>/<name>.deliver.txt.

The hooks get VELCRO_RECORDING (the audio), VELCRO_RECORD (its import line, JSON), VELCRO_TRANSCRIPT,
VELCRO_TRANSCRIPT_JSON, VELCRO_OUT and, for deliver, VELCRO_TIDIED.

Once every recording of one import (the lines that share an "import" id: one `velcro import` run) has all
its steps done, --done <command> runs once for that import, with VELCRO_IMPORT (the id), VELCRO_RECORDS (its
lines, a JSON list), VELCRO_RECORDINGS, VELCRO_TRANSCRIPTS and VELCRO_TRANSCRIPTS_JSON (one path per line, in
manifest order) and VELCRO_OUT. Each of those four is also in a file named by the same variable plus _FILE
(lists one path per line); the inline variable is left unset when it is over 64 KiB, so big imports read the
files. It runs only when every transcript's metadata names that recording and its SHA-256.
Its success is kept in the state, so it never runs twice for an import.
Lines without an import id (written by an older velcro) never run it. A step that fails (exit status
not 0, or a timeout) is tried again after 10, 20, 40 ... minutes (at most 6 hours); from the third
failure on, --alert <command> runs with one line of text last, at most once a day per file (or import) and step.
Nothing is ever deleted or changed in <dir>. One run at a time per <out> (a lock in <state>).
File variables are private copies (0600, in a 0700 temporary folder), removed after each hook;
the audio is copied and hash-checked again. VELCRO_OUT remains the output folder.
Exit status: 0 done (failures are counted in the state and retried later), 2 usage.
"""
import argparse
import datetime
import fcntl
import hashlib
import json
import os
from pathlib import Path
import secrets
import shlex
import signal
import stat
import subprocess
import sys
import tempfile
import time
import unicodedata

MANIFEST = '.velcro-import.jsonl'
BACKOFF_S, BACKOFF_MAX_S = 600, 6 * 3600
ALERT_FROM = 3
HOOK_TIMEOUT_S = 1800
STEPS = ('transcribe', 'tidy', 'deliver')
INLINE_MAX = 64 * 1024  # bytes per inline hook variable; beyond it only the *_FILE variable is set
DONE_KEY = 'import {}'  # an import's key in the failure counts, next to the file names
TRANSCRIBE = ('claude', 'codex')  # engines velcro transcribe runs


def say(*a):
    print(time.strftime('%H:%M:%S'), *a, flush=True)


def opened_path(fd):
    """Where an open file really is (every link resolved by the kernel), or None where that can't be asked."""
    try:
        if os.path.isdir('/proc/self/fd'):
            return os.readlink(f'/proc/self/fd/{fd}')
        if hasattr(fcntl, 'F_GETPATH'):  # macOS
            return os.fsdecode(fcntl.fcntl(fd, fcntl.F_GETPATH, bytes(1024)).rstrip(b'\0'))
    except OSError:
        pass
    return None


def command(line, **fill):
    """A setting's command line as argv: a leading ~ is the home folder, {name} placeholders are filled.
    Returns (argv, whether a {file} placeholder was there)."""
    argv = [os.path.expanduser(w) for w in shlex.split(line)]
    used = any('{file}' in w for w in argv)
    return [w.format_map(_Keep(fill)) for w in argv], used


class _Keep(dict):
    def __missing__(self, key):  # an unknown {word} stays as typed
        return '{' + key + '}'


def run_limited(argv, timeout, env=None):
    """subprocess.run with a time limit that stops the whole process group, so an engine's own children
    (a converter, a container client) don't outlive it."""
    p = subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
                         env=env, start_new_session=True)
    try:
        out, err = p.communicate(timeout=timeout)
    except BaseException:  # the time limit, or this run being stopped (SIGTERM, ^C): the engine goes with it
        os.killpg(p.pid, signal.SIGTERM)
        try:
            p.communicate(timeout=30)
        except subprocess.TimeoutExpired:
            os.killpg(p.pid, signal.SIGKILL)
            p.communicate()
        raise
    return subprocess.CompletedProcess(argv, p.returncode, out, err)


def put(path, data):
    """Write through a temporary name, so a half-written file never has the real name. The temporary file
    is a new one (mkstemp: never an existing name or a link someone left in a shared folder)."""
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix=f'.{path.name}.', suffix='.part')
    try:
        os.fchmod(fd, file_mode(path))
        with os.fdopen(fd, 'w') as f:
            f.write(data)
        os.replace(tmp, path)  # replaces a link at path itself, never what it points to
    except BaseException:
        Path(tmp).unlink(missing_ok=True)
        raise


def file_mode(path, dir_fd=None):
    """Only our own regular files keep their permissions; all other replacements are private."""
    try:
        st = os.stat(path, dir_fd=dir_fd, follow_symlinks=False)
        if stat.S_ISREG(st.st_mode) and st.st_uid == os.geteuid():
            return stat.S_IMODE(st.st_mode)
    except FileNotFoundError:
        pass
    return 0o600


def read_small(path, limit=1 << 20):
    """A small regular file's text, or None. The output folder may be in a share others write to: a FIFO there
    must not hang the run (it would hold the lock forever) and a link to /dev/zero must not fill the memory."""
    try:
        fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
    except OSError:
        return None
    with os.fdopen(fd, 'rb') as f:
        if not stat.S_ISREG(os.fstat(f.fileno()).st_mode):
            return None
        data = f.read(limit + 1)
    return None if len(data) > limit else data.decode('utf-8', 'replace')


def put_at(folder, name, data):
    """put() into the open folder <folder> (a descriptor): the folder can't be swapped for a link meanwhile."""
    while True:
        tmp = f'.{name}.{secrets.token_hex(4)}.part'
        try:
            fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=folder)
            break
        except FileExistsError:
            continue
    try:
        os.fchmod(fd, file_mode(name, dir_fd=folder))
        with os.fdopen(fd, 'w') as f:
            f.write(data)
        os.replace(tmp, name, src_dir_fd=folder, dst_dir_fd=folder)
    except BaseException:
        try:
            os.unlink(tmp, dir_fd=folder)
        except FileNotFoundError:
            pass
        raise


def output_path(path, boundary):
    """Resolve user aliases before entering the share, leaving every component on it literal."""
    out = Path(os.path.abspath(path))
    for _ in range(40):
        for prefix in (*reversed(out.parents), out):
            if prefix == boundary:
                return out
            if prefix.is_symlink():
                target = Path(os.path.abspath(prefix.parent / os.readlink(prefix)))
                out = target / out.relative_to(prefix)
                break
        else:
            return out
    raise RuntimeError(f'{path}: too many output folder links')


class Run:
    def __init__(self, a):
        self.a = a
        self.dir, self.out, self.state_dir = Path(a.dir).resolve(), Path(a.out), Path(a.state)
        # Find the mount before looking at output links: its siblings and final component are untrusted too.
        mount, dev = self.dir, self.dir.stat().st_dev
        while mount.parent != mount and mount.parent.stat().st_dev == dev:
            mount = mount.parent
        out, configured_dir = Path(os.path.abspath(self.out)), Path(os.path.abspath(a.dir))
        if out == configured_dir or configured_dir in out.parents:
            # The configured recordings alias is trusted once, including for the default output.
            out = self.dir / out.relative_to(configured_dir)
        self.out_path = output_path(out, mount)
        tag = hashlib.sha1(str(self.out.resolve()).encode()).hexdigest()[:10]
        self.state_file = self.state_dir / f'recordings-{tag}.json'
        self.lock_file = self.state_dir / f'recordings-{tag}.lock'
        self.load()

    def load(self):
        try:
            self.state = json.loads(self.state_file.read_text())
        except (FileNotFoundError, ValueError):
            self.state = {}

    def save(self):
        self.state_dir.mkdir(parents=True, exist_ok=True)
        self.state['out'] = str(self.out)
        put(self.state_file, json.dumps(self.state, ensure_ascii=False, indent=1))

    # ---- what to do ----

    def records(self):
        self.record_errors = []
        self.registered = set()
        index = self.dir / MANIFEST
        if not index.exists():
            self.record_errors.append({'stage': 'manifest missing'})
            return []
        recs = []
        for line in index.read_text().splitlines():
            try:
                r = json.loads(line)
            except ValueError:
                self.record_errors.append({'stage': 'invalid manifest line'})
                continue
            if not isinstance(r, dict) or not isinstance(r.get('file'), str) or not r['file']:
                self.record_errors.append({'stage': 'invalid manifest record'})
                continue
            self.registered.add(unicodedata.normalize('NFC', r['file']))
            if str(r.get('imported_at', '')) >= (self.a.since or ''):
                recs.append(r)
        # A Mac writes decomposed Hangul; the share may keep the composed name. Take the actual spelling
        # only when exactly one file matches.
        names = [p.name for p in self.dir.iterdir() if p.is_file()]
        nfc = {}
        for n in names:
            nfc.setdefault(unicodedata.normalize('NFC', n), []).append(n)
        for r in recs:
            m = nfc.get(unicodedata.normalize('NFC', r['file']), [])
            if r['file'] not in names and len(m) == 1:
                r['file'] = m[0]
        latest = {}  # the same file imported again: the last line wins
        for r in recs:
            if not self.inside(self.dir / r['file']):
                # The folder may be a share others write to: an absolute path, ../ or a link leading out of it
                # would have the engine read (and the transcript show) a file of this account's. This only
                # skips them early; copy_in() is what keeps a link swapped in later out.
                self.record_errors.append({'file': r['file'], 'stage': 'outside recordings folder'})
                if self.a.command != 'check':
                    say(f"{r['file']}: outside {self.dir}, skipped")
                continue
            latest[r['file']] = r
        return list(latest.values())

    def inside(self, path):
        """Whether path, with every link followed, is in the recordings folder."""
        top = os.path.realpath(self.dir)
        return os.path.commonpath([top, os.path.realpath(path)]) == top != os.path.realpath(path)

    def open_out(self, create=False):
        """Walk from the root using descriptors, never following a link or mkdir-ing through one.
        This also protects output elsewhere on the share's filesystem, outside recordings.dir."""
        fd = os.open(self.out_path.anchor, os.O_RDONLY | os.O_DIRECTORY)
        try:
            for part in self.out_path.parts[1:]:
                flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW
                try:
                    child = os.open(part, flags, dir_fd=fd)
                except FileNotFoundError:
                    if not create:
                        raise
                    try:
                        os.mkdir(part, dir_fd=fd)
                    except FileExistsError:
                        pass
                    child = os.open(part, flags, dir_fd=fd)
                os.close(fd)
                fd = child
            return fd
        except OSError as e:
            os.close(fd)
            raise RuntimeError(f'{self.out}: outside the output folder or not a folder: {e.strerror}') from e

    def paths(self, rec):
        # memo.wav and memo.m4a share a stem: the one transcribed first keeps memo.*, the other gets
        # memo.m4a.*, so neither is taken for done by the other's transcript.
        stem = Path(rec['file']).stem
        try:
            owner = json.loads(read_small(self.out / f'{stem}.json')).get('file')
        except (OSError, ValueError, AttributeError, TypeError):
            owner = None
        if isinstance(owner, str) and unicodedata.normalize('NFC', owner) != unicodedata.normalize('NFC', rec['file']):
            stem = rec['file']
        return {'audio': self.dir / rec['file'], 'txt': self.out / f'{stem}.txt', 'json': self.out / f'{stem}.json',
                        'tidy': self.out / f'{stem}.tidy.txt', 'deliver': self.out / f'{stem}.deliver.txt'}

    def next_step(self, rec):
        p = self.paths(rec)
        if not p['audio'].exists():
            return None, 'source gone'
        if not p['json'].exists():
            step = 'transcribe'
        elif self.a.tidy and not p['tidy'].exists():
            step = 'tidy'
        elif self.a.deliver and not p['deliver'].exists():
            step = 'deliver'
        else:
            return None, 'done'
        f = self.state.get('failures', {}).get(rec['file'])
        if f and f.get('step') == step and f.get('next', 0) > time.time():
            return None, f'{step} (try {f["count"] + 1} after {time.strftime("%m-%d %H:%M", time.localtime(f["next"]))})'
        return step, step

    # ---- steps ----

    def copy_in(self, rec, path, work):
        """Copy the recording into work, and its SHA-256. Anyone who writes to the folder can turn the file into
        a link at any moment, so the check is made on the open file itself and the engine only sees the copy."""
        real = os.path.realpath(path)
        try:  # O_NONBLOCK: a FIFO left there must not hang the run (it fails the S_ISREG check below)
            fd = os.open(real, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK)
        except OSError as e:
            raise RuntimeError(f"{rec['file']}: can't open it: {e.strerror}")
        with os.fdopen(fd, 'rb') as f:
            st = os.fstat(fd)
            where = opened_path(fd)
            if where is None:  # no /proc or F_GETPATH: the open file must be the one at the checked path
                at = os.lstat(real)
                where = real if (at.st_dev, at.st_ino) == (st.st_dev, st.st_ino) else None
            if not stat.S_ISREG(st.st_mode) or not where or not self.inside(where):
                raise RuntimeError(f"{rec['file']}: outside {self.dir} or not a file")
            (work / 'in').mkdir(mode=0o700)
            copy, h = work / 'in' / Path(rec['file']).name, hashlib.sha256()
            with os.fdopen(os.open(copy, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'wb') as out:
                for b in iter(lambda: f.read(1 << 20), b''):
                    h.update(b)
                    out.write(b)
        return copy, h.hexdigest()

    def transcribe(self, rec):
        # A private folder (0700, in the state folder, not the share) for the copy and the engine's JSON.
        self.state_dir.mkdir(parents=True, exist_ok=True)
        with tempfile.TemporaryDirectory(prefix='velcro-rec-', dir=self.state_dir) as work:
            return self.transcribe_in(rec, Path(work))

    def transcribe_in(self, rec, work):
        p = self.paths(rec)
        audio, got = self.copy_in(rec, p['audio'], work)
        if rec.get('sha256') and got != rec['sha256']:
            raise RuntimeError(f"{rec['file']}: SHA-256 doesn't match the import record")
        limit = self.a.timeout or 1800 + float(rec.get('duration_s') or 3 * 3600) * 0.5
        engine_json = work / 'engine.json'
        t0 = time.time()
        if self.a.engine in TRANSCRIBE:
            argv = [self.a.velcro, 'transcribe', '--json', '--engine', self.a.engine] + (['--lang', self.a.lang] if self.a.lang else []) + [str(audio)]
        else:
            argv, has_file = command(self.a.engine, file=str(audio), json=str(engine_json))
            if not has_file:
                argv.append(str(audio))
        try:
            r = run_limited(argv, limit)
        except subprocess.TimeoutExpired:
            raise RuntimeError(f'the engine took over {round(limit / 60)} min, stopped')
        if r.returncode:
            raise RuntimeError(f'the engine failed (exit {r.returncode}): {r.stderr.strip()[-300:]}')
        text, detail = r.stdout, None
        if self.a.engine in TRANSCRIBE:
            detail = json.loads(text)
            text = detail.pop('text', '')
            if 'file' in detail:
                detail['file'] = str(p['audio'])  # the recording, not the private copy
        elif engine_json.exists():
            try:
                detail = json.loads(engine_json.read_text())
            except ValueError:
                detail = None
        if not text.strip():
            raise RuntimeError('the engine printed no text')
        doc = {'file': rec['file'], 'sha256': rec.get('sha256'), 'duration_s': rec.get('duration_s'),
               'engine': self.a.engine if self.a.engine in TRANSCRIBE else 'command', 'command': argv[0],
               'wall_s': round(time.time() - t0), 'at': now_iso(), 'detail': detail}
        out = self.open_out()
        try:
            put_at(out, p['txt'].name, text if text.endswith('\n') else text + '\n')
            put_at(out, p['json'].name, json.dumps(doc, ensure_ascii=False, indent=1))  # last: marks the step done
        finally:
            os.close(out)
        return f"{len(text)} characters in {doc['wall_s']} s"

    def hook(self, rec, step):
        self.state_dir.mkdir(parents=True, exist_ok=True)
        with tempfile.TemporaryDirectory(prefix='velcro-rec-', dir=self.state_dir) as work:
            return self.hook_in(rec, step, Path(work))

    def copy_out(self, out, path, work):
        """A private 0600 copy in work of an output-folder file, opened without following links."""
        fd = os.open(path.name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=out)
        copy = work / path.name
        with os.fdopen(fd, 'rb') as source:
            if not stat.S_ISREG(os.fstat(fd).st_mode):
                raise RuntimeError(f'{path}: not a regular file')
            with os.fdopen(os.open(copy, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'wb') as dest:
                for b in iter(lambda: source.read(1 << 20), b''):
                    dest.write(b)
        return copy

    def hook_in(self, rec, step, work):
        p = self.paths(rec)
        audio, got = self.copy_in(rec, p['audio'], work)
        if rec.get('sha256') and got != rec['sha256']:
            raise RuntimeError(f"{rec['file']}: SHA-256 doesn't match the import record")
        env = dict(os.environ, VELCRO_RECORDING=str(audio), VELCRO_RECORD=json.dumps(rec, ensure_ascii=False),
                   VELCRO_OUT=str(self.out))
        out = self.open_out()
        try:
            for key, var in (('txt', 'VELCRO_TRANSCRIPT'), ('json', 'VELCRO_TRANSCRIPT_JSON'), ('tidy', 'VELCRO_TIDIED')):
                if key == 'tidy' and step != 'deliver':
                    continue
                try:
                    env[var] = str(self.copy_out(out, p[key], work))
                except FileNotFoundError:
                    if key == 'tidy' and not self.a.tidy:
                        env[var] = ''
                        continue
                    raise
        finally:
            os.close(out)
        argv, _ = command(getattr(self.a, step))
        try:
            r = run_limited(argv, HOOK_TIMEOUT_S, env)
        except subprocess.TimeoutExpired:
            raise RuntimeError(f'{step} took over {HOOK_TIMEOUT_S // 60} min, stopped')
        if r.returncode:
            raise RuntimeError(f'{step} failed (exit {r.returncode}): {(r.stderr or r.stdout).strip()[-300:]}')
        out = self.open_out()
        try:
            put_at(out, p[step].name, r.stdout)
        finally:
            os.close(out)
        return f'{step} ok'

    def import_groups(self, recs):
        groups = {}
        for rec in recs:
            if isinstance(rec.get('import'), str) and rec['import']:
                groups.setdefault(rec['import'], []).append(rec)
        return groups

    def revision(self, recs):
        members = sorted((r['file'], r.get('sha256')) for r in recs)
        return hashlib.sha256(json.dumps(members, ensure_ascii=False).encode()).hexdigest()

    def import_finished(self, imp, recs):
        saved = self.state.get('imports', {}).get(imp)
        return isinstance(saved, dict) and saved.get('revision') == self.revision(recs)

    def imports(self, recs):
        """Ready imports whose current set of recordings has not been handed on."""
        groups = self.import_groups(recs)
        return {i: g for i, g in groups.items()
                if not self.import_finished(i, g) and all(self.next_step(r)[1] == 'done' for r in g)}

    def waiting(self, imp):
        """When the done hook of this import failed and its next try is still ahead: a line saying so, else None."""
        f = self.state.get('failures', {}).get(DONE_KEY.format(imp))
        if f and f.get('next', 0) > time.time():
            return f'done (try {f["count"] + 1} after {time.strftime("%m-%d %H:%M", time.localtime(f["next"]))})'
        return None

    def done(self, imp, recs):
        # Same rule as tidy/deliver: the hook gets private copies, never paths on the share.
        self.state_dir.mkdir(parents=True, exist_ok=True)
        with tempfile.TemporaryDirectory(prefix='velcro-rec-', dir=self.state_dir) as work:
            return self.done_in(imp, recs, Path(work))

    def done_in(self, imp, recs, work):
        audios, txts, jsons = [], [], []
        for i, rec in enumerate(recs):
            own = work / str(i)  # one folder per recording: names may repeat across folders
            own.mkdir(mode=0o700)
            p = self.paths(rec)
            audio, got = self.copy_in(rec, p['audio'], own)
            if rec.get('sha256') and got != rec['sha256']:
                raise RuntimeError(f"{rec['file']}: SHA-256 doesn't match the import record")
            audios.append(str(audio))
            out = self.open_out()
            try:
                txt, doc = self.copy_out(out, p['txt'], own), self.copy_out(out, p['json'], own)
            finally:
                os.close(out)
            # The same test as `check`, on the copies: a transcript left from an older recording with the same
            # name must not be handed on as this one's.
            try:
                meta = json.loads(doc.read_text())
            except ValueError:
                meta = None
            if (not isinstance(meta, dict)
                    or unicodedata.normalize('NFC', meta.get('file') or '') != unicodedata.normalize('NFC', rec['file'])
                    or meta.get('sha256') != rec.get('sha256') or txt.stat().st_size == 0):
                raise RuntimeError(f"{rec['file']}: its transcript is missing or belongs to another recording")
            txts.append(str(txt))
            jsons.append(str(doc))
        # Lists and the import's lines also go in private files: a big import would not fit in the environment
        # (Linux refuses one variable over 128 KiB). The inline variables are set only when they fit.
        values = {'VELCRO_RECORDS': json.dumps(recs, ensure_ascii=False), 'VELCRO_RECORDINGS': '\n'.join(audios),
                  'VELCRO_TRANSCRIPTS': '\n'.join(txts), 'VELCRO_TRANSCRIPTS_JSON': '\n'.join(jsons)}
        env = dict(os.environ, VELCRO_IMPORT=imp, VELCRO_IMPORT_REVISION=self.revision(recs), VELCRO_OUT=str(self.out))
        for var, value in values.items():
            f = work / f'{var.lower()}.txt'
            with os.fdopen(os.open(f, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), 'w', encoding='utf-8') as w:
                w.write(value if var == 'VELCRO_RECORDS' or not value else value + '\n')  # lists: one line each
            env[var + '_FILE'] = str(f)
            if len(value.encode()) <= INLINE_MAX:
                env[var] = value
            else:
                env.pop(var, None)
        argv, _ = command(self.a.done)
        try:
            r = run_limited(argv, HOOK_TIMEOUT_S, env)
        except subprocess.TimeoutExpired:
            raise RuntimeError(f'done took over {HOOK_TIMEOUT_S // 60} min, stopped')
        if r.returncode:
            raise RuntimeError(f'done failed (exit {r.returncode}): {(r.stderr or r.stdout).strip()[-300:]}')
        return f'done ok ({len(recs)} recording{"" if len(recs) == 1 else "s"})'

    # ---- failures ----

    def failed(self, rec, step, err):
        """Count a failure of rec's step; rec is an import line, or {'file': DONE_KEY} for an import's done hook."""
        f = self.state.setdefault('failures', {}).get(rec['file'])
        if not f or f.get('step') != step:
            f = {'count': 0}
        f['count'] += 1
        f.update(step=step, error=str(err)[-300:], at=int(time.time()),
                 next=int(time.time() + min(BACKOFF_S * 2 ** (f['count'] - 1), BACKOFF_MAX_S)))
        self.state['failures'][rec['file']] = f
        self.save()
        say(f"{rec['file']}: {step} failed (try {f['count']}):", err)
        if f['count'] >= ALERT_FROM and self.a.alert:
            key, today = f"{rec['file']}:{step}", time.strftime('%Y-%m-%d')
            sent = self.state.setdefault('alerted', {})
            if sent.get(key) != today:
                text = (f"velcro recordings: {rec['file']} {step} failed {f['count']} times, next try "
                        f"{time.strftime('%m-%d %H:%M', time.localtime(f['next']))}: {str(err)[-200:]}")
                argv, _ = command(self.a.alert)
                try:
                    r = subprocess.run(argv + [text], stdin=subprocess.DEVNULL, capture_output=True, text=True, timeout=120)
                    if r.returncode == 0:
                        sent[key] = today
                        self.save()
                    else:
                        say('alert failed:', (r.stderr or r.stdout).strip()[-200:])
                except (OSError, subprocess.TimeoutExpired) as e:
                    say('alert failed:', e)

    # ---- commands ----

    def check(self):
        """Read-only reconciliation for a timer: gaps are exit 1, never 'nothing new'."""
        recs = self.records()
        issues = list(self.record_errors)
        configured = {k: bool(getattr(self.a, k)) for k in ('engine', 'tidy', 'deliver', 'done')}
        if not configured['engine']:
            issues.append({'stage': 'engine not configured'})
        for required in self.a.require:
            if not configured[required]:
                issues.append({'stage': f'{required} not configured'})
        # registered includes older lines excluded by since: those files are not orphan audio.
        unregistered = [p.name for p in self.dir.iterdir()
                        if p.is_file() and not p.name.startswith('.')
                        and p.suffix.lower().lstrip('.') in self.a.ext.split()
                        and unicodedata.normalize('NFC', p.name) not in self.registered]
        issues.extend({'file': n, 'stage': 'not in manifest'} for n in sorted(unregistered))
        complete = 0
        for rec in recs:
            if 'done' in self.a.require and not (isinstance(rec.get('import'), str) and rec['import']):
                issues.append({'file': rec['file'], 'stage': 'import id missing for done hook'})
            _, stage = self.next_step(rec)
            if stage == 'done':
                p = self.paths(rec)
                try:
                    doc = json.loads(read_small(p['json']))
                    text_stat = p['txt'].lstat()
                    valid = (isinstance(doc, dict)
                             and unicodedata.normalize('NFC', doc.get('file') or '') == unicodedata.normalize('NFC', rec['file'])
                             and doc.get('sha256') == rec.get('sha256')
                             and stat.S_ISREG(text_stat.st_mode) and text_stat.st_size > 0)
                except (ValueError, TypeError, OSError):
                    valid = False
                if not valid:
                    stage = 'transcript missing or mismatched'
            if stage == 'done':
                complete += 1
            else:
                issues.append({'file': rec['file'], 'stage': stage})
        if configured['done']:
            for imp, group in self.import_groups(recs).items():
                if not self.import_finished(imp, group):
                    issues.append({'import': imp, 'stage': self.waiting(imp) or 'done hook pending'})
        report = {'ok': not issues, 'at': now_iso(), 'configured': configured,
                  'counts': {'recordings': len(recs), 'complete': complete,
                             'pending': len(recs) - complete, 'unregistered': len(unregistered)}, 'issues': issues}
        print(json.dumps(report, ensure_ascii=False))
        return 0 if report['ok'] else 1

    def status(self):
        recs = self.records()
        if not recs:
            say('no recordings in', self.dir / MANIFEST)
        for rec in recs:
            _, what = self.next_step(rec)
            print(f"{rec['file']:28} {what}")
        if self.a.done:
            for imp in self.imports(recs):
                print(f"{DONE_KEY.format(imp):28} {self.waiting(imp) or 'done hook'}")
        return 0

    def run(self):
        self.state_dir.mkdir(parents=True, exist_ok=True)
        with open(self.lock_file, 'w') as lock:
            try:
                fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                say('another run is still going')
                return 0
            self.load()  # again, now that no other run can change it: what was read before may be stale
            return self.steps()

    def steps(self):
        started, worked = time.time(), False
        for rec in self.records():
            while True:
                step, _ = self.next_step(rec)
                if not step:
                    break
                if time.time() - started > self.a.budget:
                    say('run budget used, the rest waits for the next run')
                    return 0
                worked = True
                try:
                    out = self.open_out(create=True)
                    os.close(out)
                    msg = self.transcribe(rec) if step == 'transcribe' else self.hook(rec, step)
                except Exception as e:  # any failure is counted and retried; the next file still runs
                    self.failed(rec, step, e)
                    break
                self.state.get('failures', {}).pop(rec['file'], None)
                self.state.setdefault('done', {}).setdefault(rec['file'], {})[step] = now_iso()
                self.save()
                say(f"{rec['file']}: {msg}")
        if self.a.done:
            for imp, recs in self.imports(self.records()).items():
                if self.waiting(imp):
                    continue
                if time.time() - started > self.a.budget:
                    say('run budget used, the rest waits for the next run')
                    return 0
                worked, key = True, DONE_KEY.format(imp)
                try:
                    msg = self.done(imp, recs)
                except Exception as e:
                    self.failed({'file': key}, 'done', e)
                    continue
                self.state.get('failures', {}).pop(key, None)
                self.state.setdefault('imports', {})[imp] = {'at': now_iso(), 'revision': self.revision(recs)}
                self.save()
                say(f'{key}: {msg}')
        if not worked:
            say('nothing new')
        return 0


def now_iso():
    return datetime.datetime.now().astimezone().isoformat(timespec='seconds')


def main(argv=None):
    p = argparse.ArgumentParser(prog='velcro-recordings', description='Transcribe, tidy and deliver imported recordings.')
    p.add_argument('command', choices=('run', 'status', 'check'))
    p.add_argument('--dir', required=True, help='the folder velcro import writes to (with .velcro-import.jsonl)')
    p.add_argument('--out', required=True, help='where transcripts and hook output go')
    p.add_argument('--state', required=True, help='folder for the failure counts and the lock')
    p.add_argument('--engine', required=True, help='a command line ({file}, {json}), or claude or codex')
    p.add_argument('--require', action='append', choices=('tidy', 'deliver', 'done'), default=[],
                   help='check: fail if this hook is not configured (repeatable)')
    p.add_argument('--ext', default='wav mp3 m4a aac flac wma ogg', help='audio extensions checked against the manifest')
    p.add_argument('--velcro', default='velcro', help='the velcro command, for --engine claude or codex')
    p.add_argument('--lang', default='', help='language for --engine claude or codex, e.g. ko')
    p.add_argument('--tidy', default='', help='command run after transcribing')
    p.add_argument('--deliver', default='', help='command run after tidying')
    p.add_argument('--done', default='', help='command run once per import, when all its recordings are done')
    p.add_argument('--alert', default='', help='command run with a line of text from the third failure on')
    p.add_argument('--since', default='', help='only recordings imported at or after this time (2026-10-21T00:00)')
    p.add_argument('--timeout', type=lambda v: float(v) if v else 0, default=0,
                   help='seconds the engine may take (default 1800 + half the recording)')
    p.add_argument('--budget', type=float, default=40 * 60, help='seconds after which no new step starts (2400)')
    a = p.parse_args(argv)
    if not Path(a.dir).is_dir():
        print(f'velcro-recordings: no such folder: {a.dir}', file=sys.stderr)
        return 2
    r = Run(a)
    return r.run() if a.command == 'run' else r.check() if a.command == 'check' else r.status()


if __name__ == '__main__':
    # Stopped by a timer or kill: unwind (removing the private copy, stopping the engine) instead of vanishing.
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(143))
    sys.exit(main())
