Add resumable deep permutation passes for unmatched functions

ROADMAP.md

@@ -893,3 +893,9 @@ agent briefs and deep-search output. Planned: compiler-specific diagnostic flags
 opt-in diagnostic build command for imported projects (their ordinary build diagnostics remain available).
 Reuse discovery batches fingerprint names/cache reads and persists a separate cheap-discovery cursor; at most
 half the remaining shared reuse time goes to that phase, leaving time for wider matching.
+
+Project settings now offer a resumable deep permutation pass across unmatched functions in ROM and imported
+projects. Each function's completed bounded turn is persisted with its source/build scope and budget. Unfinished
+turns resume after restart; changed inputs become eligible again. Busy functions and candidates without scored C
+wait without being marked complete. The pass uses the same local/random searches, claims and judges as a single
+function, with a visible progress bar and stop button. No paid helper is launched.

src/decompnet/app/deep_permuter.py

@@ -49,6 +49,8 @@ def __init__(self, net, minutes, max_variants, log, fn=None):
         self.baseline_measured = False
         self.phase = 'local repairs'
         self.measuring_baseline = False
+        self.record_improvement = lambda score: None
+        self.recorded_best = None
 
     def check(self):
         self.log.check_cancel()
@@ -67,6 +69,9 @@ def percent(score):
                 'best_score': self.best, 'best_percent': percent(self.best), 'improvements': self.improvements}
 
     def report(self):
+        if self.best is not None and (self.recorded_best is None or self.best < self.recorded_best):
+            self.record_improvement(self.best)
+            self.recorded_best = self.best
         self.log.progress({'phase': self.phase, 'tested': self.tested, 'skipped': self.skipped,
                            **self.metrics(), 'variant_limit': self.limit, 'rejected': self.rejected,
                            'seconds_left': max(0, round(self.deadline-time.monotonic()))})
@@ -121,7 +126,13 @@ def _score(self, agent, proj, fn, source, best, fork):
 def run(net, slug, name, minutes, max_variants, log):
     """Visit the supported local repairs, then spend the remaining budget on random permutations."""
     proj, fn = validate(net, slug, name, minutes, max_variants)
+    sweeps = getattr(net, 'deep_sweeps', None)
+    if sweeps:
+        from ..candidate_scope import key
+        build_scope = key(net, proj, fn)
     search = Search(net, minutes, max_variants, log, fn)
+    if sweeps:
+        search.record_improvement = lambda score: sweeps.improvement(fn, score)
     search.report()
     state = State()
     agent, _ = net.local_agent('permuter')
@@ -145,17 +156,29 @@ def run(net, slug, name, minutes, max_variants, log):
                 reason = 'Matched'
             else:
                 best = baseline
-                source = fn['best_source']
-                rows = getattr(net, '_last_diffs', {}).get((agent.id, fn['id']), ([], ''))[0]
-                choices = draft_repairs.variants(source, fn['name'], proj['arch'], fn['asm'] or '', flags, limit=None)
-                choices += stack_padding_repairs.grow(source, fn['name'], proj['arch'], rows)
-                choices += [s for _,s in permuter_local.variants(source, fn['name'])]
-                for candidate in dict.fromkeys(choices):
-                    score = search._score(agent, proj, fn, candidate, best, fork)
-                    if score is not None and score < best:
-                        best = score
-                    if score == 0:
-                        reason = 'Matched'
+                seen = {fn['best_source']}
+                repair_fn = fn
+                while best > 0:
+                    search.check()
+                    source = repair_fn['best_source']
+                    rows = getattr(net, '_last_diffs', {}).get((agent.id, repair_fn['id']), ([], ''))[0]
+                    choices = draft_repairs.variants(source, repair_fn['name'], proj['arch'], repair_fn['asm'] or '', flags, limit=None)
+                    choices += stack_padding_repairs.grow(source, repair_fn['name'], proj['arch'], rows)
+                    choices += [s for _,s in permuter_local.variants(source, repair_fn['name'])]
+                    for candidate in dict.fromkeys(choices):
+                        if candidate in seen:
+                            continue
+                        seen.add(candidate)
+                        score = search._score(agent, proj, repair_fn, candidate, best, fork)
+                        if score == 0:
+                            reason = 'Matched'
+                        if score is not None and score < best:
+                            best = score
+                            # Chain repairs from the saved winner and its new diff; keep the same
+                            # deadline/compile budget and avoid retesting candidates from earlier rounds.
+                            repair_fn = net._function(name, slug)
+                            break
+                    else:
                         break
         finally:
             search._let_go(agent, fn)
@@ -179,5 +202,8 @@ def run(net, slug, name, minutes, max_variants, log):
         reason = str(error)
     search.report()
     log(f'{reason}: {search.tested} candidates tested, {search.skipped} cached misses skipped, {search.rejected} invalid calls skipped. Verified improvements are saved immediately.')
-    return {'function': name, 'reason': reason, 'tested': search.tested, 'skipped': search.skipped,
-            'rejected': search.rejected, 'matched': net._function(name, slug)['status'] == 'matched', **search.metrics()}
+    result = {'function': name, 'reason': reason, 'tested': search.tested, 'skipped': search.skipped,
+              'rejected': search.rejected, 'matched': net._function(name, slug)['status'] == 'matched', **search.metrics()}
+    if sweeps:
+        sweeps.completed(slug, fn, minutes, max_variants, result, build_scope)
+    return result

src/decompnet/app/deep_sweep.py

@@ -0,0 +1,224 @@
+"""Durable project-wide deep passes; each candidate gets a bounded turn through the normal judges."""
+import hashlib
+import json
+import threading
+import time
+
+from .. import candidate_scope, scoring
+from ..network import NetworkError
+from . import agent_tokens, deep_permuter
+
+POLL_SECONDS = 15
+SCHEMA = '''CREATE TABLE IF NOT EXISTS deep_permutation_passes(
+ function_id INTEGER PRIMARY KEY, scope TEXT NOT NULL, state TEXT NOT NULL,
+ minutes INTEGER NOT NULL, max_variants INTEGER NOT NULL, completed_at REAL,
+ result TEXT NOT NULL DEFAULT '{}', error TEXT NOT NULL DEFAULT '')'''
+
+
+def signature(net, proj, fn, minutes, variants):
+    return hashlib.sha256(repr(('deep-pass-v1', agent_tokens.target_scope(net, fn, ''),
+                               fn['best_source'], minutes, variants)).encode()).hexdigest()
+
+
+class Sweeps:
+    def __init__(self, net, jobs):
+        self.net, self.jobs = net, jobs
+        self.active = {}
+        self.lock = threading.RLock()
+        with net.store.tx() as db:
+            db.execute(SCHEMA)
+            db.execute('''CREATE TABLE IF NOT EXISTS deep_permutation_improvements(
+                function_id INTEGER PRIMARY KEY, starting_score INTEGER NOT NULL,
+                starting_percent REAL, new_score INTEGER NOT NULL,
+                new_percent REAL, updated_at REAL NOT NULL)''')
+
+    def config(self, slug):
+        proj = self.net.project_row(slug)
+        row = self.net.store.one('SELECT value FROM settings WHERE key=?', f'deep-sweep:{proj["id"]}')
+        return json.loads(row['value']) if row else {'enabled': False, 'minutes': 5, 'max_variants': 10000}
+
+    def save(self, slug, cfg):
+        proj = self.net.project_row(slug)
+        with self.net.store.tx() as db:
+            db.execute('INSERT OR REPLACE INTO settings(key,value) VALUES(?,?)',
+                       (f'deep-sweep:{proj["id"]}', json.dumps(cfg)))
+
+    def history(self, slug, fn):
+        row = self.net.store.one('SELECT * FROM deep_permutation_passes WHERE function_id=?', fn['id'])
+        if not row:
+            return {'state': 'not tried', 'completed': False}
+        result = dict(row)
+        result['stale'] = row['scope'] != signature(self.net, self.net.project_row(slug), fn,
+                                                   row['minutes'], row['max_variants'])
+        result['completed'] = row['state'] == 'completed' and not result['stale']
+        result['result'] = json.loads(row['result'])
+        return result
+
+    def improvement(self, before, score):
+        """Keep only scores measured by this search, including wins before cancellation."""
+        start = before['best_score']
+        if score is None or start is None or score >= start:
+            return
+        def percent(value):
+            if value == 0:
+                return 100.0
+            if not before['n_insns']:
+                return None
+            return max(0.0, min(99.9 if value else 100.0,
+                               100.0 * (1 - value / max(1, before['n_insns'] * 100))))
+        with self.net.store.tx() as db:
+            db.execute('''INSERT INTO deep_permutation_improvements
+                (function_id,starting_score,starting_percent,new_score,new_percent,updated_at)
+                VALUES(?,?,?,?,?,?) ON CONFLICT(function_id) DO UPDATE SET
+                new_score=excluded.new_score,new_percent=excluded.new_percent,updated_at=excluded.updated_at
+                WHERE excluded.new_score < deep_permutation_improvements.new_score''',
+                (before['id'], start, percent(start), score, percent(score), time.time()))
+
+    def improvements(self, project_id):
+        rows = self.net.store.all('''SELECT i.*, f.name AS function FROM deep_permutation_improvements i
+            JOIN functions f ON f.id=i.function_id WHERE f.project_id=? ORDER BY i.updated_at DESC,f.id''', project_id)
+        return [dict(row) | {'starting_percent_text': scoring.percent_text(row['starting_percent']),
+                             'new_percent_text': scoring.percent_text(row['new_percent'])} for row in rows]
+
+    def status(self, slug):
+        proj = self.net.project_row(slug)
+        counts = self.net.store.one("SELECT COUNT(*) total, SUM(best_source IS NOT NULL AND best_score IS NOT NULL) ready "
+                                    "FROM functions WHERE project_id=? AND status='open'", proj['id'])
+        completed = self.net.store.one("SELECT COUNT(*) n FROM deep_permutation_passes d JOIN functions f ON f.id=d.function_id "
+                                      "WHERE f.project_id=? AND d.state='completed'", proj['id'])['n']
+        errors = self.net.store.one("SELECT COUNT(*) n FROM deep_permutation_passes d JOIN functions f ON f.id=d.function_id WHERE f.project_id=? AND d.state='error'", proj['id'])['n']
+        improvements = self.improvements(proj['id'])
+        with self.lock:
+            current = dict(self.active.get(slug, {}))
+        job = self.jobs.jobs.get(current.get('job'), {})
+        return self.config(slug) | {'total': counts['total'], 'ready': counts['ready'] or 0,
+                                   'improved': len(improvements), 'improvements': improvements,
+                                   'completed': completed, 'errors': errors, 'state': job.get('state', 'idle'), **current,
+                                   'progress': job.get('progress', {}), 'error': job.get('error')}
+
+    def start(self, slug, minutes=5, max_variants=10000, repeat=False):
+        if (isinstance(minutes, bool) or not isinstance(minutes, int) or not 1 <= minutes <= 120 or
+                isinstance(max_variants, bool) or not isinstance(max_variants, int) or not 1 <= max_variants <= 1000000):
+            raise NetworkError('Choose 1–120 minutes and 1–1,000,000 variants per function.')
+        self.net.project_row(slug)
+        if slug in self.net.links:
+            raise NetworkError('Run the deep pass on the room host’s project.')
+        with self.lock:
+            previous = self.active.get(slug, {})
+            if self.jobs.jobs.get(previous.get('job'), {}).get('state') == 'running':
+                raise NetworkError('A deep pass is already running for this project. Stop it before starting another.')
+            if repeat:
+                with self.net.store.tx() as db:
+                    db.execute("DELETE FROM deep_permutation_passes WHERE function_id IN "
+                               "(SELECT id FROM functions WHERE project_id=? AND status='open')",
+                               self.net.project_row(slug)['id'])
+            self.save(slug, {'enabled': True, 'minutes': minutes, 'max_variants': max_variants})
+            job = self.jobs.start('deep-permutation-pass', lambda log: self.run(slug, minutes, max_variants, log),
+                                  cancel_message='Deep pass stopped. Verified improvements and completed passes are saved.')
+            self.active[slug] = {'job': job}
+        return self.status(slug)
+
+    def stop(self, slug):
+        with self.lock:
+            cfg = self.config(slug)
+            self.save(slug, cfg | {'enabled': False})
+            job = self.active.get(slug, {}).get('job')
+            if job:
+                self.jobs.cancel(job)
+        return self.status(slug)
+
+    def resume(self):
+        # An interrupted function is retried; only completed bounded turns are skipped.
+        for proj in self.net.store.all('SELECT slug FROM projects'):
+            cfg = self.config(proj['slug'])
+            if cfg['enabled']:
+                try:
+                    self.start(proj['slug'], cfg['minutes'], cfg['max_variants'])
+                except NetworkError as error:
+                    self.save(proj['slug'], cfg | {'enabled': False, 'resume_error': str(error)})
+                    self.jobs.logger.warning('Deep pass could not resume: %s', error)
+
+    def shutdown(self):
+        # Preserve enabled intent for the next launch, but let current searches release claims.
+        with self.lock:
+            for current in self.active.values():
+                self.jobs.cancel(current['job'])
+
+    def completed(self, slug, before, minutes, variants, result, build_scope):
+        self.improvement(before, result.get('best_score'))
+        fresh = self.net._function(before['name'], slug)
+        proj = self.net.project_row(slug)
+        stable = build_scope == candidate_scope.key(self.net, proj, before)
+        self.record(fresh, signature(self.net, proj, fresh, minutes, variants),
+                    'completed' if stable else 'pending', minutes, variants, result)
+
+    def record(self, fn, scope, state, minutes, variants, result=None, error=''):
+        with self.net.store.tx() as db:
+            db.execute('INSERT OR REPLACE INTO deep_permutation_passes '
+                       '(function_id,scope,state,minutes,max_variants,completed_at,result,error) VALUES(?,?,?,?,?,?,?,?)',
+                       (fn['id'], scope, state, minutes, variants, time.time() if state == 'completed' else None,
+                        json.dumps(result or {}), error))
+
+    def run(self, slug, minutes, variants, log):
+        """Finish one turn per unchanged candidate; wait for claims or usable C without marking them done."""
+        visited = tested = matched = 0
+        while True:
+            log.check_cancel()
+            proj = self.net.project_row(slug)
+            pending = waiting = 0
+            rows = self.net.store.all("SELECT * FROM functions WHERE project_id=? AND status='open' ORDER BY size,id", proj['id'])
+            for row in rows:
+                log.check_cancel()
+                fn = self.net._function(row['name'], slug)
+                if fn['status'] != 'open':
+                    continue
+                if not fn['best_source'] or fn['best_score'] is None:
+                    waiting += 1
+                    continue
+                scope = signature(self.net, proj, fn, minutes, variants)
+                previous = self.net.store.one('SELECT * FROM deep_permutation_passes WHERE function_id=?', fn['id'])
+                if previous and previous['scope'] == scope and previous['state'] in ('completed', 'error'):
+                    continue
+                pending += 1
+                if self.net.store.one('SELECT 1 FROM claims WHERE function_id=? AND expires_at>?', fn['id'], time.time()):
+                    waiting += 1
+                    continue
+                build_scope = candidate_scope.key(self.net, proj, fn)
+                self.record(fn, scope, 'running', minutes, variants)
+                log(f'Deep pass: {fn["name"]} · up to {minutes} minutes / {variants:,} variants')
+                total = len(rows)
+                def progress(value, name=fn['name'], done=visited, wins=matched, total=total, blocked=waiting):
+                    log.progress({'function': name, 'functions_tested': done, 'matches': wins,
+                                  'total_unmatched': total, 'waiting': blocked} | value)
+                def function_log(line):
+                    log(line)
+                function_log.check_cancel = log.check_cancel
+                function_log.progress = progress
+                try:
+                    result = deep_permuter.run(self.net, slug, fn['name'], minutes, variants, function_log)
+                    # Also records test backends and older search providers that do not publish history.
+                    # The actual single-function search uses the same completion gate.
+                    self.completed(slug, fn, minutes, variants, result, build_scope)
+                    visited += 1
+                    tested += result['tested']
+                    matched += int(result['matched'])
+                except NetworkError as error:
+                    # Claim races are temporary. A bad unchanged candidate stays visible as an error,
+                    # not completed, and a changed source/header or an explicit repeat retries it.
+                    busy = self.net.store.one('SELECT 1 FROM claims WHERE function_id=? AND expires_at>?', fn['id'], time.time())
+                    self.record(fn, scope, 'pending' if busy or 'claim' in str(error).lower() or str(error).startswith('This function is working') else 'error', minutes, variants, error=str(error))
+                    log(f'{fn["name"]}: {error}')
+                except BaseException:
+                    self.record(fn, scope, 'pending', minutes, variants)
+                    raise
+                progress({'phase': 'moving to next function', 'tested': tested, 'functions_tested': visited, 'matches': matched})
+            if not pending and not waiting:
+                self.save(slug, self.config(slug) | {'enabled': False})
+                log('Deep pass finished. Completed bounded passes are saved; this is not an exhaustive search of all C.')
+                return {'functions_tested': visited, 'tested': tested, 'matched': matched}
+            log.progress({'phase': 'waiting for unclaimed functions or scored C', 'functions_tested': visited,
+                          'matches': matched, 'waiting': waiting, 'total_unmatched': len(rows)})
+            # Cancellation remains responsive while another worker owns the remaining functions.
+            for _ in range(POLL_SECONDS):
+                log.check_cancel()
+                time.sleep(1)