"""Consigliere supervisor: local, one-pass, fail-closed management runtime.
No trading, tax submission, autonomous staffing or guaranteed message delivery.
"""
import argparse
import contextlib
import fcntl
import hashlib
import hmac
import json
import os
from pathlib import Path
import re
import signal
import sqlite3
import stat
import subprocess
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
from datetime import datetime, timezone, timedelta
from decimal import Decimal, ROUND_HALF_UP, localcontext

VERSION = 1
MAX_CENTS = 900000000000000000
STATES = ('PENDING_EXECUTION','STAGED','DEPLOYED','VERIFIED_CLEARED','REPORTED_CLEARED','FAILED','UNKNOWN')


def now():
    return datetime.now(timezone.utc)


def deadline(value):
    if not isinstance(value, str) or len(value) > 80:
        raise ValueError('Invalid deadline')
    result = datetime.fromisoformat(value.replace('Z', '+00:00'))
    if result.tzinfo is None or result.utcoffset() is None:
        raise ValueError('Deadline requires timezone offset')
    return result.astimezone(timezone.utc)


def cents(value, legacy=False):
    if legacy and isinstance(value, float):
        value = str(value)
    if type(value) not in (str, int) or len(str(value)) > 64:
        raise ValueError('Money must be a bounded decimal string or integer')
    if not re.fullmatch(r'[0-9]+(?:\.[0-9]+)?', str(value)):
        raise ValueError('Invalid money format')
    d = Decimal(str(value))
    if not d.is_finite() or d < 0 or d > Decimal(MAX_CENTS) / 100:
        raise ValueError('Money outside range')
    if not legacy and d.as_tuple().exponent < -2:
        raise ValueError('Sub-cent precision unsupported for USD budgets')
    with localcontext() as ctx:
        ctx.prec = 80
        return int(d.quantize(Decimal('.01'), rounding=ROUND_HALF_UP) * 100)


def canonical(payload):
    copied = dict(payload)
    headers = dict(payload['security_headers'])
    headers.pop('signature', None)
    copied['security_headers'] = headers
    return json.dumps(copied, sort_keys=True, separators=(',', ':'), ensure_ascii=False, allow_nan=False).encode('utf-8')


def no_duplicates(pairs):
    result = {}
    for k, v in pairs:
        if k in result:
            raise ValueError('Duplicate JSON key')
        result[k] = v
    return result


def load_json(raw):
    return json.loads(raw, object_pairs_hook=no_duplicates, parse_constant=lambda _: (_ for _ in ()).throw(ValueError('Nonfinite JSON')))


def private_file(path):
    path = Path(path)
    if path.is_symlink():
        raise ValueError('Symlink rejected')
    st = path.stat()
    if not stat.S_ISREG(st.st_mode) or st.st_uid != os.getuid() or st.st_mode & 0o077:
        raise ValueError('File must be owned by the current user and private (600 or 400)')
    return path


def private_json(path):
    return load_json(private_file(path).read_text())


@contextlib.contextmanager
def transaction(conn):
    conn.execute('BEGIN IMMEDIATE')
    try:
        yield
        conn.commit()
    except BaseException:
        conn.rollback()
        raise


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        return None


def push(message, credentials):
    if not credentials or not credentials.get('app_token') or not credentials.get('user_token'):
        return 'NOT_CONFIGURED'
    if len(message) > 1024:
        return 'PERMANENT_REJECTION'
    body = urllib.parse.urlencode({'token': credentials['app_token'], 'user': credentials['user_token'], 'message': message, 'priority': 1, 'sound': 'siren'}).encode()
    req = urllib.request.Request('https://api.pushover.net/1/messages.json', data=body, method='POST')
    try:
        with urllib.request.build_opener(NoRedirect).open(req, timeout=8) as response:
            data = load_json(response.read(16385))
            if response.status == 200 and isinstance(data, dict) and type(data.get('status')) is int and data['status'] == 1:
                return 'API_ACCEPTED'
            return 'PERMANENT_REJECTION'
    except urllib.error.HTTPError as e:
        if e.code == 429:
            return 'RETRY'
        if 400 <= e.code < 500:
            return 'PERMANENT_REJECTION'
        return 'UNKNOWN'
    except Exception:
        # Server may have accepted before timeout: never blindly retransmit.
        return 'UNKNOWN'


class ConsigliereProductionCore:
    def __init__(self, db_filepath, monitor_dir, *, sender_keys=None, handlers=None, notifier=None, notification_configured=False):
        self.db_path = Path(db_filepath).absolute()
        self.monitor_dir = Path(monitor_dir).absolute()
        self.sender_keys = sender_keys or {}
        self.handlers = handlers or {}
        self.notifier = notifier
        self.notification_configured = notification_configured
        self.worker_id = str(uuid.uuid4())
        self.conn = None
        self.lock = None
        for directory in (self.db_path.parent, self.monitor_dir):
            directory.mkdir(parents=True, exist_ok=True, mode=0o700)
            if directory.is_symlink():
                raise ValueError('Symlink directory rejected')
        # All instances of this runtime use one OS lock; crash releases it.
        lock_path = str(self.db_path) + '.supervisor.lock'
        fd = os.open(lock_path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
        self.lock = os.fdopen(fd, 'a+')
        lock_stat = os.fstat(fd)
        if lock_stat.st_uid != os.getuid() or lock_stat.st_mode & 0o077:
            self.lock.close(); self.lock = None
            raise ValueError('Unsafe lock file permissions')
        try:
            fcntl.flock(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
            if self.db_path.exists():
                private_file(self.db_path)
            if not self.db_path.exists():
                fd = os.open(self.db_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
                os.close(fd)
            self.conn = sqlite3.connect(self.db_path, timeout=5, isolation_level=None)
            os.chmod(self.db_path, 0o600)
            self.conn.row_factory = sqlite3.Row
            self.conn.execute('PRAGMA foreign_keys=ON')
            self.conn.execute('PRAGMA synchronous=FULL')
            self._migrate()
            # Lock proves prior runtime is gone; side effects may already have happened.
            with transaction(self.conn):
                self.conn.execute("UPDATE remediation_lifecycle SET execution_status='UNKNOWN' WHERE execution_status='STAGED'")
                self.conn.execute("UPDATE notification_outbox SET delivery_status='UNKNOWN' WHERE delivery_status='CLAIMED'")
        except BaseException:
            self.close()
            raise

    def _snapshot(self):
        target = self.db_path.with_name(self.db_path.name + '.backup-' + str(uuid.uuid4()) + '.sqlite3')
        fd = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
        os.close(fd)
        dst = sqlite3.connect(target)
        try:
            self.conn.backup(dst)
            if [r[0] for r in dst.execute('PRAGMA integrity_check')] != ['ok']:
                raise ValueError('Backup integrity failure')
            # Verify accessible schema and data in the snapshot; retained for recovery.
            for r in dst.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").fetchall():
                dst.execute('SELECT COUNT(*) FROM "' + r[0].replace('"', '""') + '"').fetchone()
        finally:
            dst.close()
            os.chmod(target, 0o600)
        return target

    def _migrate(self):
        c = self.conn
        version = c.execute('PRAGMA user_version').fetchone()[0]
        existing = c.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'").fetchall()
        if version == VERSION:
            self._verify_schema()
            return
        if version != 0:
            raise ValueError('Unsupported database version; no changes made')
        names = {r[0] for r in existing}
        allowed = {'enterprise_portfolio','remediation_lifecycle','autonomous_resolution_rules','ingestion_journal','notification_outbox'}
        if names - allowed:
            raise ValueError('Unknown tables require a reviewed migration; no changes made')
        if c.execute("SELECT 1 FROM sqlite_master WHERE type IN ('view','trigger') LIMIT 1").fetchone():
            raise ValueError('Custom views/triggers require a reviewed migration')
        if names:
            self._snapshot()
        c.execute('PRAGMA foreign_keys=OFF')
        try:
            with transaction(c):
                # Lock established before reading migration inputs.
                sequence_rows = c.execute('SELECT name,seq FROM sqlite_sequence').fetchall() if c.execute("SELECT 1 FROM sqlite_master WHERE name='sqlite_sequence'").fetchone() else []
                originals = {n: [dict(r) for r in c.execute('SELECT * FROM "' + n + '"')] for n in names}
                allowed_fields = {
                    'enterprise_portfolio': {'project_id','project_name','industry_vertical','lifecycle_state','supervisor_deadline','allocated_budget','allocated_budget_cents','days_lagging_count','logged_at'},
                    'remediation_lifecycle': {'remediation_id','associated_project','trigger_condition','deployed_directive','execution_evidence_log','execution_status','triggered_at','migration_audit_notes'},
                    'autonomous_resolution_rules': {'rule_id','target_vertical','trigger_condition','pre_approved_action'},
                    'ingestion_journal': {'event_uuid','payload_hash_sha256','processed_at'},
                    'notification_outbox': {'notification_id','associated_project','alert_payload_text','retry_count','delivery_status','queued_at','retry_backoff_expiry','worker_owner_id'},
                }
                for n in names:
                    cols = {r[1] for r in c.execute('PRAGMA table_info("' + n + '")')}
                    if cols - allowed_fields[n]:
                        raise ValueError('Unknown legacy columns require review')
                for n in ('notification_outbox','ingestion_journal','autonomous_resolution_rules','remediation_lifecycle','enterprise_portfolio'):
                    if n in names:
                        c.execute('DROP TABLE "' + n + '"')
                self._create_tables()
                expected = {}
                for row in originals.get('enterprise_portfolio', []):
                    amount = row.get('allocated_budget_cents')
                    if amount is None:
                        amount = cents(row['allocated_budget'], legacy=True)
                    if type(amount) is not int or not 0 <= amount <= MAX_CENTS:
                        raise ValueError('Invalid legacy budget')
                    r = {k: row[k] for k in ('project_id','project_name','industry_vertical','lifecycle_state','supervisor_deadline','logged_at')}
                    r['allocated_budget_cents'] = amount
                    self._insert('enterprise_portfolio', r)
                    expected.setdefault('enterprise_portfolio', []).append(r)
                for row in originals.get('remediation_lifecycle', []):
                    r = dict(row)
                    r.setdefault('trigger_condition', 'UNKNOWN_TRIGGER')
                    r.setdefault('execution_evidence_log', None)
                    r.setdefault('migration_audit_notes', None)
                    r['incident_id'] = 'legacy-' + str(row['remediation_id'])
                    r['worker_owner_id'] = None
                    self._insert('remediation_lifecycle', r)
                    expected.setdefault('remediation_lifecycle', []).append(r)
                for row in originals.get('autonomous_resolution_rules', []):
                    r = dict(row, enabled=0)
                    self._insert('autonomous_resolution_rules', r)
                    expected.setdefault('autonomous_resolution_rules', []).append(r)
                for row in originals.get('ingestion_journal', []):
                    r = dict(row)
                    r.setdefault('payload_hash_sha256', None)
                    self._insert('ingestion_journal', r)
                    expected.setdefault('ingestion_journal', []).append(r)
                for row in originals.get('notification_outbox', []):
                    r = dict(row)
                    r.setdefault('retry_backoff_expiry', None)
                    r.setdefault('worker_owner_id', None)
                    r['delivery_status'] = {'SENT_VERIFIED':'LEGACY_UNVERIFIED','DEVICE_ACKNOWLEDGED':'LEGACY_UNVERIFIED','CLAIMED':'UNKNOWN'}.get(r['delivery_status'], r['delivery_status'])
                    self._insert('notification_outbox', r)
                    expected.setdefault('notification_outbox', []).append(r)
                # Preserve every source field, including discarded lag metrics, in migration audit.
                for n, rows in originals.items():
                    c.execute('INSERT INTO migration_audit(source_table,source_rows_json,migrated_at) VALUES(?,?,?)', (n, json.dumps(rows, sort_keys=True, allow_nan=False), now().isoformat()))
                for n, rows in expected.items():
                    actual = [dict(r) for r in c.execute('SELECT * FROM "' + n + '"')]
                    if sorted(map(lambda x: json.dumps(x, sort_keys=True), rows)) != sorted(map(lambda x: json.dumps(x, sort_keys=True), actual)):
                        raise ValueError('Full-field migration reconciliation failed')
                for name, value in sequence_rows:
                    if name in ('enterprise_portfolio','remediation_lifecycle','autonomous_resolution_rules','notification_outbox'):
                        current = c.execute('SELECT seq FROM sqlite_sequence WHERE name=?',(name,)).fetchone()
                        if current:
                            c.execute('UPDATE sqlite_sequence SET seq=? WHERE name=?',(max(value,current[0]),name))
                        else:
                            c.execute('INSERT INTO sqlite_sequence(name,seq) VALUES(?,?)',(name,value))
                self._verify_schema()
                c.execute('PRAGMA user_version=1')
        finally:
            c.execute('PRAGMA foreign_keys=ON')
            if c.execute('PRAGMA foreign_keys').fetchone()[0] != 1:
                raise ValueError('Foreign keys unavailable')

    def _create_tables(self):
        statements = [
            "CREATE TABLE enterprise_portfolio(project_id INTEGER PRIMARY KEY AUTOINCREMENT,project_name TEXT NOT NULL UNIQUE,industry_vertical TEXT NOT NULL,lifecycle_state TEXT NOT NULL CHECK(lifecycle_state IN ('ACTIVE','INACTIVE','HOLD')),supervisor_deadline TEXT NOT NULL,allocated_budget_cents INTEGER NOT NULL CHECK(typeof(allocated_budget_cents)='integer' AND allocated_budget_cents BETWEEN 0 AND 900000000000000000),logged_at TEXT NOT NULL)",
            "CREATE TABLE remediation_lifecycle(remediation_id INTEGER PRIMARY KEY AUTOINCREMENT,associated_project TEXT NOT NULL REFERENCES enterprise_portfolio(project_name),trigger_condition TEXT NOT NULL,deployed_directive TEXT NOT NULL,execution_evidence_log TEXT,execution_status TEXT NOT NULL CHECK(execution_status IN ('PENDING_EXECUTION','STAGED','DEPLOYED','VERIFIED_CLEARED','REPORTED_CLEARED','FAILED','UNKNOWN')),triggered_at TEXT NOT NULL,migration_audit_notes TEXT,incident_id TEXT NOT NULL UNIQUE,worker_owner_id TEXT)",
            "CREATE UNIQUE INDEX idx_open_remediations ON remediation_lifecycle(associated_project,trigger_condition) WHERE execution_status IN ('STAGED','DEPLOYED','PENDING_EXECUTION','UNKNOWN','FAILED')",
            "CREATE TABLE autonomous_resolution_rules(rule_id INTEGER PRIMARY KEY AUTOINCREMENT,target_vertical TEXT NOT NULL,trigger_condition TEXT NOT NULL,pre_approved_action TEXT NOT NULL,enabled INTEGER NOT NULL DEFAULT 0 CHECK(enabled IN (0,1)),UNIQUE(target_vertical,trigger_condition))",
            "CREATE TABLE ingestion_journal(event_uuid TEXT PRIMARY KEY NOT NULL,payload_hash_sha256 TEXT,processed_at TEXT NOT NULL)",
            "CREATE TABLE notification_outbox(notification_id INTEGER PRIMARY KEY AUTOINCREMENT,associated_project TEXT NOT NULL,alert_payload_text TEXT NOT NULL,retry_count INTEGER NOT NULL DEFAULT 0 CHECK(retry_count>=0),delivery_status TEXT NOT NULL CHECK(delivery_status IN ('PENDING','CLAIMED','GATEWAY_ACCEPTED','FAILED','UNKNOWN','LEGACY_UNVERIFIED')),queued_at TEXT NOT NULL,retry_backoff_expiry TEXT,worker_owner_id TEXT)",
            "CREATE TABLE migration_audit(audit_id INTEGER PRIMARY KEY,source_table TEXT NOT NULL,source_rows_json TEXT NOT NULL,migrated_at TEXT NOT NULL)",
            "CREATE TABLE project_event_versions(project_name TEXT PRIMARY KEY REFERENCES enterprise_portfolio(project_name),issued_at TEXT NOT NULL)",
            "CREATE TABLE audit_events(audit_id INTEGER PRIMARY KEY,event_type TEXT NOT NULL,reference TEXT,recorded_at TEXT NOT NULL)",
        ]
        for sql in statements:
            self.conn.execute(sql)

    def _verify_schema(self):
        required = {'enterprise_portfolio': {'project_id','project_name','allocated_budget_cents','logged_at'}, 'remediation_lifecycle': {'incident_id','worker_owner_id','execution_status'}, 'autonomous_resolution_rules': {'enabled'}, 'ingestion_journal': {'payload_hash_sha256'}, 'notification_outbox': {'worker_owner_id','retry_backoff_expiry','delivery_status'}, 'migration_audit': {'source_rows_json'}, 'audit_events': {'event_type'}}
        for n, fields in required.items():
            if not fields <= {r[1] for r in self.conn.execute('PRAGMA table_info("' + n + '")')}:
                raise ValueError('Schema verification failed: ' + n)
        if self.conn.execute('PRAGMA foreign_key_check').fetchall():
            raise ValueError('Foreign-key integrity failure')
        if [r[0] for r in self.conn.execute('PRAGMA integrity_check')] != ['ok']:
            raise ValueError('Database integrity failure')
        # Compare exact runtime-owned schema to an independently created baseline.
        actual_conn = self.conn
        baseline = sqlite3.connect(':memory:')
        try:
            self.conn = baseline
            self._create_tables()
            expected = {r[0]:r[1] for r in baseline.execute("SELECT name,sql FROM sqlite_master WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'")}
        finally:
            self.conn = actual_conn
            baseline.close()
        actual = {r[0]:r[1] for r in actual_conn.execute("SELECT name,sql FROM sqlite_master WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'")}
        if actual != expected:
            raise ValueError('Schema constraints differ from the supported version')

    def _insert(self, table, row):
        keys = list(row)
        self.conn.execute('INSERT INTO ' + table + '(' + ','.join(keys) + ') VALUES(' + ','.join('?' for _ in keys) + ')', [row[k] for k in keys])

    def _audit(self, kind, reference=None):
        self.conn.execute('INSERT INTO audit_events(event_type,reference,recorded_at) VALUES(?,?,?)', (kind, reference, now().isoformat()))

    def _archive(self, path, category):
        folder = self.monitor_dir / category
        folder.mkdir(mode=0o700, exist_ok=True)
        if folder.is_symlink():
            raise ValueError('Symlink archive directory rejected')
        target = folder / (str(uuid.uuid4()) + '.json')
        os.rename(path, target)
        os.chmod(target, 0o600)

    def autonomous_directory_sync_scan(self):
        results = []
        for path in sorted(self.monitor_dir.glob('*.json')):
            outcome = 'REJECTED'
            event = None
            try:
                if path.is_symlink() or not path.is_file() or path.stat().st_size > 1048576:
                    raise ValueError('Invalid input file')
                payload = load_json(path.read_text(encoding='utf-8'))
                headers = payload['security_headers']
                event = str(uuid.UUID(headers['event_uuid']))
                raw_key = self.sender_keys.get(headers['sender_id'])
                if not raw_key or len(raw_key) < 32:
                    raise ValueError('Sender not configured')
                signature = headers['signature']
                if not isinstance(signature, str) or not re.fullmatch('[0-9a-f]{64}', signature):
                    raise ValueError('Invalid signature format')
                data = canonical(payload)
                if not hmac.compare_digest(hmac.new(raw_key, data, hashlib.sha256).hexdigest(), signature):
                    raise ValueError('Invalid signature')
                digest = hashlib.sha256(data).hexdigest()
                meta = payload['metadata']
                for key in ('project_name','vertical','lifecycle_state'):
                    if not isinstance(meta[key], str) or not 1 <= len(meta[key]) <= 120:
                        raise ValueError('Invalid metadata')
                d = deadline(meta['deadline']).isoformat()
                amount = cents(meta['budget_value'])
                with transaction(self.conn):
                    old = self.conn.execute('SELECT payload_hash_sha256 FROM ingestion_journal WHERE event_uuid=?', (event,)).fetchone()
                    if old:
                        outcome = 'DUPLICATE' if old[0] == digest else 'CONFLICT'
                    else:
                        issued = deadline(headers['issued_at'])
                        if abs((now() - issued).total_seconds()) > 86400:
                            raise ValueError('New event expired or future-dated')
                        previous = self.conn.execute('SELECT issued_at FROM project_event_versions WHERE project_name=?',(meta['project_name'],)).fetchone()
                        if previous and issued <= deadline(previous[0]):
                            raise ValueError('Out-of-order project event')
                        self.conn.execute('INSERT INTO ingestion_journal VALUES(?,?,?)', (event, digest, now().isoformat()))
                        self.conn.execute("INSERT INTO enterprise_portfolio(project_name,industry_vertical,lifecycle_state,supervisor_deadline,allocated_budget_cents,logged_at) VALUES(?,?,?,?,?,?) ON CONFLICT(project_name) DO UPDATE SET industry_vertical=excluded.industry_vertical,lifecycle_state=excluded.lifecycle_state,supervisor_deadline=excluded.supervisor_deadline,allocated_budget_cents=excluded.allocated_budget_cents,logged_at=excluded.logged_at", (meta['project_name'],meta['vertical'],meta['lifecycle_state'],d,amount,now().isoformat()))
                        self.conn.execute('INSERT INTO project_event_versions(project_name,issued_at) VALUES(?,?) ON CONFLICT(project_name) DO UPDATE SET issued_at=excluded.issued_at',(meta['project_name'],issued.isoformat()))
                        outcome = 'INGESTED'
                    self._audit(outcome, event)
            except Exception:
                with transaction(self.conn):
                    self._audit('REJECTED', None)
            try:
                self._archive(path, 'processed' if outcome in ('INGESTED','DUPLICATE') else 'quarantine')
            except Exception:
                with transaction(self.conn):
                    self._audit('ARCHIVE_PENDING', event)
            results.append(outcome)
        return results

    def _handler(self, action):
        config = self.handlers.get(action)
        if not config:
            raise ValueError('Handler not configured')
        path = Path(config['path'])
        private_file(path)
        if path.resolve() != path.absolute() or path.parent.name != 'actions':
            raise ValueError('Handler path not trusted')
        parent = path.parent.stat()
        if parent.st_uid != os.getuid() or parent.st_mode & 0o022:
            raise ValueError('Handler directory writable by others')
        if not os.access(path, os.X_OK):
            raise ValueError('Handler not executable')
        digest = hashlib.sha256(path.read_bytes()).hexdigest()
        if not hmac.compare_digest(digest, config['sha256']):
            raise ValueError('Handler hash mismatch')
        return path

    def execute_autonomous_triage_loop(self, *, execute=False):
        results = []
        rows = self.conn.execute("SELECT * FROM enterprise_portfolio WHERE lifecycle_state='ACTIVE'").fetchall()
        for project in rows:
            try:
                due = deadline(project['supervisor_deadline'])
            except ValueError:
                self._audit('INVALID_DEADLINE', project['project_name'])
                continue
            if now() <= due:
                continue
            rule = self.conn.execute("SELECT * FROM autonomous_resolution_rules WHERE target_vertical=? AND trigger_condition='DEADLINE_LAG' AND enabled=1", (project['industry_vertical'],)).fetchone()
            if not rule:
                self._audit('NO_APPROVED_RULE', project['project_name'])
                continue
            if not execute:
                results.append('DRY_RUN')
                continue
            try:
                handler = self._handler(rule['pre_approved_action'])
            except Exception:
                self._audit('HANDLER_BLOCKED', project['project_name'])
                results.append('BLOCKED')
                continue
            incident = str(uuid.uuid4())
            try:
                with transaction(self.conn):
                    cur = self.conn.execute("INSERT INTO remediation_lifecycle(associated_project,trigger_condition,deployed_directive,execution_status,triggered_at,incident_id,worker_owner_id) VALUES(?,'DEADLINE_LAG',?,'STAGED',?,?,?)", (project['project_name'],rule['pre_approved_action'],now().isoformat(),incident,self.worker_id))
                    rid = cur.lastrowid
            except sqlite3.IntegrityError:
                results.append('ALREADY_OPEN')
                continue
            status = 'UNKNOWN'
            evidence = 'Process launch or result uncertain; operator reconciliation required'
            try:
                with tempfile.TemporaryFile() as output:
                    proc = subprocess.Popen([str(handler)], cwd=handler.parent, stdout=output, stderr=output, start_new_session=True, env={'PATH':'/usr/bin:/bin','QZOTIC_INCIDENT_ID':incident})
                    try:
                        code = proc.wait(timeout=10)
                        status = 'DEPLOYED' if code == 0 else 'FAILED'
                        evidence = 'Process exit code: ' + str(code) + '; business outcome remains unverified'
                    except subprocess.TimeoutExpired:
                        os.killpg(proc.pid, signal.SIGKILL)
                        proc.wait()
                        evidence = 'Process timed out; partial side effects may exist'
                # Never retain raw command output; it can contain private data.
            except Exception:
                pass
            with transaction(self.conn):
                self.conn.execute('UPDATE remediation_lifecycle SET execution_status=?,execution_evidence_log=? WHERE remediation_id=? AND worker_owner_id=? AND execution_status=\'STAGED\'', (status,evidence,rid,self.worker_id))
                text = 'QZOTIC: ' + project['project_name'] + '; action ' + rule['pre_approved_action'] + '; status ' + status + '. Business outcome unverified.'
                self.conn.execute("INSERT INTO notification_outbox(associated_project,alert_payload_text,delivery_status,queued_at) VALUES(?,?,'PENDING',?)", (project['project_name'],text,now().isoformat()))
                self._audit('ACTION_' + status, incident)
            results.append(status)
        return results

    def process_notification_retry_outbox(self, *, send=False):
        if not send or not self.notification_configured or not self.notifier:
            return ['NOT_CONFIGURED' if send else 'DRY_RUN']
        results = []
        for row in self.conn.execute("SELECT * FROM notification_outbox WHERE delivery_status='PENDING' ORDER BY notification_id LIMIT 10").fetchall():
            if row['retry_backoff_expiry'] and deadline(row['retry_backoff_expiry']) > now():
                continue
            nid = row['notification_id']
            with transaction(self.conn):
                changed = self.conn.execute("UPDATE notification_outbox SET delivery_status='CLAIMED',worker_owner_id=? WHERE notification_id=? AND delivery_status='PENDING'", (self.worker_id,nid)).rowcount
            if not changed:
                continue
            try:
                answer = self.notifier(row['alert_payload_text'])
            except Exception:
                answer = 'UNKNOWN'
            attempts = row['retry_count']
            expiry = None
            if answer == 'NOT_CONFIGURED':
                state = 'PENDING'
            elif answer == 'API_ACCEPTED':
                state = 'GATEWAY_ACCEPTED'; attempts += 1
            elif answer == 'RETRY':
                attempts += 1
                state = 'FAILED' if attempts >= 5 else 'PENDING'
                expiry = (now() + timedelta(seconds=30 * 2**min(attempts,5))).isoformat()
            elif answer == 'PERMANENT_REJECTION':
                state = 'FAILED'; attempts += 1
            else:
                state = 'UNKNOWN'; attempts += 1
            with transaction(self.conn):
                self.conn.execute('UPDATE notification_outbox SET delivery_status=?,retry_count=?,retry_backoff_expiry=?,worker_owner_id=NULL WHERE notification_id=? AND worker_owner_id=? AND delivery_status=\'CLAIMED\'', (state,attempts,expiry,nid,self.worker_id))
                self._audit('NOTIFICATION_' + state, str(nid))
            results.append(state)
        return results

    def render_executive_control_scoreboard(self):
        units = [dict(r) for r in self.conn.execute('SELECT * FROM enterprise_portfolio ORDER BY project_name')]
        for unit in units:
            unit['budget_usd'] = format(Decimal(unit['allocated_budget_cents']) / 100, '.2f')
            last = self.conn.execute('SELECT execution_status,execution_evidence_log,migration_audit_notes FROM remediation_lifecycle WHERE associated_project=? ORDER BY remediation_id DESC LIMIT 1', (unit['project_name'],)).fetchone()
            unit['remediation'] = dict(last) if last else None
        return {'units':units,'active_count':sum(r['lifecycle_state']=='ACTIVE' for r in units),'total_units':len(units),'budget_total_usd':format(Decimal(sum(r['allocated_budget_cents'] for r in units))/100,'.2f'),'notice':'Budgets are allocations, not assets. Historical status is not fresh assurance. This is a one-pass supervisor.'}

    def close(self):
        if self.conn:
            self.conn.close(); self.conn = None
        if self.lock:
            self.lock.close(); self.lock = None

    def __enter__(self):
        return self

    def __exit__(self, *args):
        self.close()


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--root', type=Path, default=Path('/Users/bpa/ForesixaQuant'))
    parser.add_argument('--execute-actions', action='store_true')
    parser.add_argument('--send-notifications', action='store_true')
    args = parser.parse_args()
    os.umask(0o077)
    root = args.root.absolute()
    keys_path = root / 'sender_keys.json'
    key_data = private_json(keys_path) if keys_path.exists() else {}
    keys = {}
    for sender, value in key_data.items():
        if not isinstance(value,str) or not re.fullmatch('[0-9a-f]{64,128}',value) or len(value)%2:
            raise ValueError('Sender key must be 32-64 random bytes encoded as hex')
        keys[sender] = bytes.fromhex(value)
    handlers_path = root / 'approved_handlers.json'
    handlers = private_json(handlers_path) if handlers_path.exists() else {}
    creds_path = root / 'notification_credentials.json'
    creds = private_json(creds_path) if creds_path.exists() else None
    with ConsigliereProductionCore(root/'qzotic_pure_sandbox.db',root/'enterprise_logs',sender_keys=keys,handlers=handlers,notifier=lambda msg:push(msg,creds),notification_configured=bool(creds and creds.get('app_token') and creds.get('user_token'))) as core:
        print(json.dumps({'ingestion':core.autonomous_directory_sync_scan(),'actions':core.execute_autonomous_triage_loop(execute=args.execute_actions),'notifications':core.process_notification_retry_outbox(send=args.send_notifications),'scoreboard':core.render_executive_control_scoreboard()},indent=2))


if __name__ == '__main__':
    main()