#!/usr/bin/env python3
"""Row 121 (#1615) mutants, round 2. Each mutant edits one or more source files, runs
the named test set, and restores every file from .orig with a sha256 check.

Rules (Sage, sage-0332):
- every test file a set names must exist, or the run stops before mutating:
  node --test skips a missing path without failing;
- a control run of each set on the unmutated source must pass with 0 skipped
  before any mutant runs, so a kill can't be a test that already failed;
- each edit's pattern must occur exactly once;
- a schema.sql mutant also rewrites SCHEMA_SHA256 in store.mjs to the mutated
  file's digest, so the store loads and the kill is a test, not the load check;
- round 2: a schema.sql edit also makes the same replace, once, in
  migrations/3d-3e.sql, or the import replay of the migration chain throws
  migration-target-mismatch and the load, not a test, makes the kill.
"""
import hashlib, os, re, shutil, signal, subprocess, sys

WT = os.path.expanduser('~/darkwing-scratch/r121/r2wt')
PKT = os.path.expanduser('~/darkwing-scratch/r121/pkt2')
TMP = os.path.expanduser('~/darkwing-scratch/r121/tmp')

SETS = {
    'SCHEMA': ('packages/bus', ['tests/threads.test.mjs', 'tests/schema.test.mjs', 'tests/messaging.test.mjs']),
    'BUS': ('packages/bus', ['tests/threads.test.mjs', 'tests/messaging.test.mjs', 'tests/end-launch.test.mjs',
                             'tests/schema.test.mjs', 'tests/store.test.mjs', 'tests/process.test.mjs']),
    'SEAT': ('packages/seat', ['tests/session.test.mjs', 'tests/sandbox.test.mjs']),
    'CLI': ('packages/cli', ['tests/thread-launch.test.mjs', 'tests/launcher.test.mjs', 'tests/launch-limits.test.mjs']),
    'BUSINESS': ('packages/business', ['tests/business.test.mjs', 'tests/chief-of-staff.test.mjs']),
    'MIGRATION': ('packages/bus', ['tests/thread-migration.test.mjs', 'tests/comment-migration.test.mjs',
                                   'tests/chief-of-staff.test.mjs']),
}

SCHEMA = 'packages/bus/schema.sql'
STORE = 'packages/bus/src/store.mjs'
MIG3E = 'packages/bus/src/migrations/3d-3e.sql'
MIGRATE_CLI = 'packages/bus/src/migrate-schema.mjs'
BROKER = 'packages/bus/src/broker.mjs'
PROCESS = 'packages/bus/src/process.mjs'
SESSION = 'packages/seat/src/session.mjs'
SANDBOX = 'packages/seat/src/sandbox.mjs'
LAUNCHER = 'packages/cli/src/launcher.mjs'
BUSINESS = 'packages/business/src/business.mjs'

TRIGGERS = [
    'threads_project_business', 'thread_one_holder', 'thread_one_per_run', 'thread_release_by_holder',
    'thread_release_same_run', 'thread_revoke_needs_decision', 'thread_claims_business',
    'message_threads_match', 'delivery_same_thread',
] + [f'{t}_no_{op}' for t in ('projects', 'threads', 'thread_claims', 'message_threads', 'thread_opens')
     for op in ('update', 'delete', 'replace')]

# (name, set, [(file, old, new), ...])
CODE = [
    # broker.mjs: #threadFor, #thread, bindLaunch, role.claim, receive, thread.create, threadsState
    ('b-human-must-use-thread-send', 'BUS', [(BROKER, "    if (s.human) fail('thread-required-use-thread-send');\n", '')]),
    ('b-instance-mismatch', 'BUS', [(BROKER, ".instance !== to) fail('thread-instance-mismatch');", ".instance !== to && false) fail('thread-instance-mismatch');")]),
    ('b-chain-one-step', 'BUS', [(BROKER, '      at = m?.in_reply_to ?? null;\n', '      at = null;\n')]),
    ('b-chain-depth-constant', 'BUS', [(BROKER, 'at !== null && !seen.has(at); )', 'at !== null && !seen.has(at) && seen.size < 3; )')]),
    ('b-no-thread-returns-null', 'BUS', [(BROKER, "    fail('thread-required-reply-to-thread-message');\n", '    return null;\n')]),
    ('b-thread-not-owned', 'BUS', [(BROKER, "    if (t.owner !== s.human) fail('thread-not-owned');\n", '')]),
    ('b-bind-thread-required', 'BUS', [(BROKER, "      if (record.thread === undefined) fail('thread-required');\n", '')]),
    ('b-bind-thread-not-found', 'BUS', [(BROKER, "        fail('thread-not-found');\n    } else if (record.thread", "        ;\n    } else if (record.thread")]),
    ('b-bind-thread-not-allowed', 'BUS', [(BROKER, "    } else if (record.thread !== undefined) fail('thread-not-allowed');\n", '    }\n')]),
    ('b-thread-already-held', 'BUS', [(BROKER, "          if (this.#threadHolder(s.thread)?.op === 'claim') fail('thread-already-held');\n", '')]),
    ('b-receive-any-thread', 'BUS', [
        (BROKER, "            (s.thread ? ' AND EXISTS(SELECT 1 FROM message_threads mt WHERE mt.message=m.id AND mt.thread=?)' : '') +\n", ''),
        (BROKER, '          ...(s.thread ? [s.thread] : []),\n', ''),
    ]),
    ('b-thread-instance-missing', 'BUS', [(BROKER, "        if (!instances.length) fail('thread-instance-missing');\n", '')]),
    ('b-thread-instance-ambiguous', 'BUS', [(BROKER, "        if (instances.length > 1) fail('thread-instance-ambiguous');\n", '')]),
    ('b-state-pending-zero', 'CLI', [(BROKER, '        pending: r.pending,\n', '        pending: 0,\n')]),
    ('b-state-last-open-null', 'BUS', [(BROKER, '        lastOpen: r.last_open,\n', '        lastOpen: null,\n')]),
    ('b-state-holder-null', 'CLI', [(BROKER, "        holder: r.claim_op === 'claim' ? r.holder_run : null,\n", '        holder: null,\n')]),
    ('p-threads-state-op', 'BUS', [(PROCESS, "  if (m.op === 'threadsState') return runtime.threadsState(m.business);\n", '')]),
    # schema.sql: delivery_same_thread scoped to messages to the thread's instance (28556 change 1)
    ('s-delivery-unscoped', 'SCHEMA', [(SCHEMA, '      WHERE mt.message = NEW.message AND m.to_role = t.instance\n', '      WHERE mt.message = NEW.message\n')]),
    # seat
    ('seat-segment-dotdot', 'SEAT', [(SESSION, ' && ID_PATTERN.test(x) && !x.includes("..");', ' && ID_PATTERN.test(x);')]),
    ('seat-segment-pattern', 'SEAT', [(SESSION, ' && ID_PATTERN.test(x) && !x.includes("..");', ' && !x.includes("..");')]),
    ('seat-store-dir-check', 'SEAT', [(SESSION, '    if (st.isSymbolicLink() || !st.isDirectory() || st.uid !== process.getuid() || st.mode & 0o077) throw', '    if (false) throw')]),
    ('seat-store-mode-check', 'SEAT', [(SESSION, ' || st.uid !== process.getuid() || st.mode & 0o077) throw', ' || st.uid !== process.getuid()) throw')]),
    ('seat-registry-thread-key', 'SEAT', [(SESSION, ' && s.instance === instance && (s.thread ?? null) === thread) ?? null;', ' && s.instance === instance) ?? null;')]),
    ('seat-no-thread-binds', 'SEAT', [(SANDBOX, '  if (threadStore) {\n', '  if (false) {\n')]),
    # launcher
    ('l-check-thread-required', 'CLI', [(LAUNCHER, '    if (threaded && !thread) throw new Refusal("thread-required", `${instance} runs only for a thread`);\n', '')]),
    ('l-check-thread-not-allowed', 'CLI', [(LAUNCHER, '    if (!threaded && thread) throw new Refusal("thread-not-allowed", `${instance} isn\'t a thread instance`);\n', '')]),
    ('l-no-thread-store', 'CLI', [(LAUNCHER, '        store = threadStore(dataRoot, b.id, thread);\n', '        store = null;\n')]),
    ('l-session-in-run-dir', 'CLI', [(LAUNCHER, '    const sessionDir = store ? store.session : join(runDir, "session");\n    if (!store) mkdirSync',
                                      '    const sessionDir = join(runDir, "session");\n    if (true) mkdirSync')]),
    ('l-idle-fixed-900', 'CLI', [(LAUNCHER, ' >= b.launch.threadIdleSeconds * 1000', ' >= 900 * 1000')]),
    ('l-open-never-starts', 'CLI', [(LAUNCHER, '        const opened = ms(t.lastOpen) > ms(t.lastClaim);\n', '        const opened = false;\n')]),
    ('l-never-start', 'CLI', [(LAUNCHER, '        if (t.holder || !(t.pending > 0 || opened)) continue;\n', '        continue;\n')]),
    ('l-start-while-held', 'CLI', [(LAUNCHER, '        if (t.holder || !(t.pending > 0 || opened)) continue;\n', '        if (!(t.pending > 0 || opened)) continue;\n')]),
    ('l-no-signature-guard', 'CLI', [(LAUNCHER, '        if (startedFor.get(t.thread) === signature) continue;\n', '')]),
    ('l-refusal-every-tick', 'CLI', [(LAUNCHER, '            if (refusedThreads.get(t.thread) === code) return;\n', '')]),
    # business
    ('biz-idle-not-checked', 'BUSINESS', [(BUSINESS, ': requirePositiveInt(launch.threadIdleSeconds, `${where}.threadIdleSeconds`);', ': launch.threadIdleSeconds;')]),
    ('biz-idle-default', 'BUSINESS', [(BUSINESS, 'launch.threadIdleSeconds === undefined ? THREAD_IDLE_SECONDS :', 'launch.threadIdleSeconds === undefined ? 60 :')]),
]

R2 = [
    # F1: thread.send runs row 120's messaging check
    ('b-thread-send-unchecked', 'BUS', [(BROKER, "    this.#checkMessaging(s.business, this.#sender(s), t.instance);\n", '')]),
    # F3: threadIdle rechecks, scopes and refuses
    ('b-idle-no-recheck', 'BUS', [(BROKER, '    if (t.pending || t.unread) return', '    if (false) return')]),
    ('b-idle-ignores-pending', 'BUS', [(BROKER, '    if (t.pending || t.unread) return', '    if (t.unread) return')]),
    ('b-idle-ignores-unread', 'BUS', [(BROKER, '    if (t.pending || t.unread) return', '    if (t.pending) return')]),
    ('b-idle-no-holder-check', 'BUS', [(BROKER, "    if (t?.holder !== record.run) fail('not-holder');\n", '')]),
    ('b-idle-no-ended-check', 'BUS', [(BROKER, "    if (this.#ended.has(key)) fail('run-ended');\n    // A run holds at most one thread", '    // A run holds at most one thread')]),
    ('b-unread-any-holder', 'BUS', [(BROKER, " AND d.op='delivered' AND d.holder_run=c.holder_run)", " AND d.op='delivered')")]),
    ('b-unread-no-claim-scope', 'BUS', [(BROKER, "            AND c.op='claim' AND EXISTS(", '            AND EXISTS(')]),
    ('p-thread-idle-op', 'BUS', [(PROCESS, "  if (m.op === 'threadIdle') return runtime.threadIdle(m.record);\n", '')]),
    ('l-idle-ignores-released', 'CLI', [(LAUNCHER, '            if (!idle.released) continue;\n', '')]),
    ('l-never-idle-stop', 'CLI', [(LAUNCHER, '          if (!stopping.has(s.run) && now - last', '          if (false && now - last')]),
]

# F5: the 3d→3e step parks pre-3e messages and reports them.
# No mutant for 'read' in the report's IN list: the broker marks read only after delivered, so it is equivalent.
R2_F5 = [
    ('m-report-counts-delivered', 'MIGRATION', [(STORE, "FROM messages m WHERE NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op IN ('delivered','read'))", "FROM messages m WHERE NOT EXISTS(SELECT 1 FROM deliveries d WHERE d.message=m.id AND d.op IN ('read'))")]),
    ('m-report-counts-all', 'MIGRATION', [(STORE, 'FROM messages m WHERE NOT EXISTS(', 'FROM messages m WHERE 1 OR NOT EXISTS(')]),
    ('m-report-by-role-only', 'MIGRATION', [(STORE, 'GROUP BY business, to_role ORDER BY', 'GROUP BY to_role ORDER BY')]),
    ('m-report-dropped', 'MIGRATION', [(STORE, '        this.#report.push(...(step.report?.(this) ?? []));\n', '')]),
    ('m-cli-no-report', 'MIGRATION', [(MIGRATE_CLI, "    for (const line of report) process.stdout.write(line + '\\n');\n", '')]),
    ('m-times-missing-table', 'MIGRATION', [(STORE, "      if (!this.get(\"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?\", table)) continue;\n", '')]),
]

CODE += R2 + R2_F5

def sha(p):
    return hashlib.sha256(open(p, 'rb').read()).hexdigest()

def sweep():
    left = []
    for pid in filter(str.isdigit, os.listdir('/proc')):
        try:
            if os.readlink(f'/proc/{pid}/cwd').startswith(TMP + '/') or os.readlink(f'/proc/{pid}/cwd') == TMP:
                left.append(int(pid)); os.kill(int(pid), signal.SIGKILL)
        except OSError:
            pass
    return left

def run_set(key):
    cwd, files = SETS[key]
    missing = [f for f in files if not os.path.isfile(os.path.join(WT, cwd, f))]
    if missing:
        raise SystemExit(f'STOP {key}: missing test files {missing}')
    p = subprocess.Popen(['node', '--test', *files], cwd=os.path.join(WT, cwd), stdout=subprocess.PIPE,
                         stderr=subprocess.STDOUT, env=env, start_new_session=True, text=True)
    try:
        out, _ = p.communicate(timeout=900)
    except subprocess.TimeoutExpired:
        os.killpg(p.pid, signal.SIGKILL); out, _ = p.communicate(); out += '\nTIMEOUT\n'
    tally = dict(re.findall(r'^ℹ (tests|pass|fail|skipped|cancelled) (\d+)$', out, re.M))
    fails = sorted({l.strip()[2:] for l in out.splitlines() if l.strip().startswith('✖ ') and l.strip() != '✖ failing tests:'})
    return p.returncode, tally, fails, out

def mirrored(edits):
    out = []
    for f, old, new in edits:
        out.append((f, old, new))
        if f == SCHEMA:
            out.append((MIG3E, old, new))
    return out

def schema_edit(name):
    src = open(SCHEMA).read()
    m = re.search(r'CREATE TRIGGER ' + re.escape(name) + r' .*?END;', src, re.S)
    assert m and src.count(f'CREATE TRIGGER {name} ') == 1, name
    body = m.group(0)
    raises = re.findall(r"SELECT RAISE\(ABORT, '(?:[^']|'')*'\)", body)
    assert len(raises) == 1, (name, raises)
    return [(SCHEMA, body, body.replace(raises[0], 'SELECT NULL'))]

os.chdir(WT)
env = dict(os.environ, TMPDIR=TMP, NO_COLOR='1')
os.makedirs(os.path.join(PKT, 'mutants'), exist_ok=True)
log = open(os.path.join(PKT, 'mutants-run.log'), 'a')
def say(*a):
    print(*a, file=log, flush=True); print(*a, flush=True)

check = '--check' in sys.argv
only = set(a for a in sys.argv[1:] if a != '--check')
mutants = [(n, k, mirrored(e)) for n, k, e in [(f't-{t}', 'SCHEMA', schema_edit(t)) for t in TRIGGERS] + CODE]
if only:
    mutants = [m for m in mutants if m[0] in only]
    assert len(mutants) == len(only), 'unknown mutant name'

# Patterns first, all of them, before anything runs.
for name, _, edits in mutants:
    for f, old, _ in edits:
        n = open(f).read().count(old)
        if n != 1:
            raise SystemExit(f'STOP {name}: pattern occurs {n} times in {f}')

if check:
    raise SystemExit(f'patterns ok: {len(mutants)} mutants')

base = {f: sha(f) for _, _, edits in mutants for f, _, _ in edits}
base[STORE] = sha(STORE)

say(f'== control {os.popen("date -u +%H:%M:%S").read().strip()} base {os.popen("git rev-parse --short HEAD").read().strip()}')
for key in sorted({s for _, s, _ in mutants}):
    rc, tally, fails, out = run_set(key)
    open(os.path.join(PKT, 'mutants', f'control-{key}.log'), 'w').write(out)
    say(f'control {key}: rc {rc} {tally}')
    if rc != 0 or tally.get('skipped', '0') != '0' or tally.get('fail') != '0':
        raise SystemExit(f'STOP control {key} not clean')

for name, key, edits in mutants:
    files = sorted({f for f, _, _ in edits})
    if SCHEMA in files:
        files.append(STORE)
    for f in files:
        shutil.copy2(f, f + '.orig')
    try:
        for f, old, new in edits:
            src = open(f).read()
            open(f, 'w').write(src.replace(old, new))
        if SCHEMA in files:
            st = open(STORE).read()
            old = re.search(r"export const SCHEMA_SHA256 = '[0-9a-f]{64}';", st).group(0)
            assert st.count(old) == 1
            open(STORE, 'w').write(st.replace(old, f"export const SCHEMA_SHA256 = '{sha(SCHEMA)}';"))
        rc, tally, fails, out = run_set(key)
        open(os.path.join(PKT, 'mutants', f'mutant-{name}.log'), 'w').write(out)
        load = 'schema-source-mismatch' in out
        verdict = 'SURVIVED' if rc == 0 else ('LOAD-FAIL' if load else 'killed')
        say(f'{name} [{key}]: {verdict} (rc {rc}, pass {tally.get("pass", "?")}, fail {tally.get("fail", "?")})')
        for l in fails:
            say(f'    fail: {l}')
        left = sweep()
        if left:
            say(f'    swept {len(left)} leftover process(es) under TMPDIR: {left}')
    finally:
        for f in files:
            shutil.move(f + '.orig', f)
            assert sha(f) == base[f], f'{f} not restored'
say('restored:', all(sha(f) == h for f, h in base.items()), os.popen('date -u +%H:%M:%S').read().strip())