diff --git a/src/infermatrix_copilot/kb_service/init_feature_discovery.py b/src/infermatrix_copilot/kb_service/init_feature_discovery.py index daaca788..c285f401 100644 --- a/src/infermatrix_copilot/kb_service/init_feature_discovery.py +++ b/src/infermatrix_copilot/kb_service/init_feature_discovery.py @@ -5,7 +5,7 @@ """ from __future__ import annotations -from collections import Counter +from collections import Counter, deque from concurrent.futures import ThreadPoolExecutor, as_completed from contextlib import contextmanager from copy import deepcopy @@ -503,60 +503,92 @@ def review(self): self.save() if stopped: return False - # Failed candidates repair separately; missing evidence remains unknown. - for key, original in sorted(self.state["candidates"].items()): - decision = self.state["reviews"].get(key, {}) - # Rewriting the same document cannot establish an implementation. - # Retain its independent verdict without paying for futile retries. - if decision.get("repair_blocked") or not any(ref["path"] in self.index.production for ref in original["evidence"]): - continue - repair_base = self.state.get("repairs", {}).get(key, {}).get("base_sha256", _hash(original)) - while decision.get("supported") != "yes" and decision.get("attempts", 0) < MAX_ATTEMPTS: - attempt = decision.get("attempts", 0) + 1 - try: - drafts = self.state.setdefault("repair_drafts", {}) - draft = drafts.get(key) + return self._repair_parallel(baseline) + + def _repair_extract(self, key, original, decision, baseline): + """One immutable extraction result; only the coordinator checkpoints it.""" + files = self._context([original]) + payload = {"repository": self.repository, "pin": self.pin, "round": "repair", + "baseline": _bounded_baseline(baseline, _json(original), [key, original.get("related_id")]), + "owners": self.owners, "candidate": original, + "files": files, "repair_reason": decision.get("reason", "")} + reply = self.call("generator", SYSTEM_DISCOVER, _prompt(payload), validate_candidates) + proof = receipt(reply) + revised = next((r for r in reply.data["candidates"] if isinstance(r, dict) and r.get("id") == key), None) + if revised is None: + raise ValueError("repair must preserve the candidate ID") + validate_candidate(revised) + from .feature_discovery_index import validate_evidence + if not all(validate_evidence(self.index, ref) and _offered_reference(self.index, files, ref) + for ref in revised["evidence"]): + raise ValueError("repair cites evidence outside its offered context") + revised = deepcopy(revised) + revised["generator_receipts"] = original["generator_receipts"] + [proof] + revised["origin_rounds"] = original["origin_rounds"] + return revised + + def _repair_parallel(self, baseline): + # A candidate has at most one native call in flight. Extract and judge + # share the same ceiling, with every draft saved before its judge starts. + def eligible(key): + row, decision = self.state["candidates"][key], self.state["reviews"].get(key, {}) + return decision.get("supported") != "yes" and decision.get("attempts", 0) < MAX_ATTEMPTS \ + and not decision.get("repair_blocked") \ + and any(ref["path"] in self.index.production for ref in row["evidence"]) + + ready = deque(key for key in sorted(self.state["candidates"]) if eligible(key)) + stopped = False + with ThreadPoolExecutor(max_workers=self.concurrency) as pool: + pending = {} + while ready or pending: + while not stopped and ready and len(pending) < self.concurrency: + key = ready.popleft() + original = deepcopy(self.state["candidates"][key]) + decision = deepcopy(self.state["reviews"].get(key, {})) + attempt = decision.get("attempts", 0) + 1 + repair_base = self.state.get("repairs", {}).get(key, {}).get("base_sha256", _hash(original)) + draft = self.state.get("repair_drafts", {}).get(key) if draft and draft["base_sha256"] == _hash(original) and draft["attempt"] == attempt: revised = deepcopy(draft["candidate"]) + phase = "judge" + future = pool.submit(self._review, [revised], baseline) else: - files = self._context([original]) - payload = {"repository": self.repository, "pin": self.pin, "round": "repair", - "baseline": _bounded_baseline(baseline, _json(original), [key, original.get("related_id")]), - "owners": self.owners, "candidate": original, - "files": files, "repair_reason": decision.get("reason", "")} - reply = self.call("generator", SYSTEM_DISCOVER, _prompt(payload), validate_candidates) - proof = receipt(reply) - revised = next((r for r in reply.data["candidates"] if isinstance(r, dict) and r.get("id") == key), None) - if revised is None: - raise ValueError("repair must preserve the candidate ID") - validate_candidate(revised) - from .feature_discovery_index import validate_evidence - if not all(validate_evidence(self.index, ref) and _offered_reference(self.index, files, ref) - for ref in revised["evidence"]): - raise ValueError("repair cites evidence outside its offered context") - revised = deepcopy(revised) - revised["generator_receipts"] = original["generator_receipts"] + [proof] - revised["origin_rounds"] = original["origin_rounds"] - drafts[key] = {"base_sha256": _hash(original), "attempt": attempt, "candidate": deepcopy(revised)} - self.save() # Successful extraction survives a later review interruption. - result = self._review([revised], baseline)[key] - self.state["candidates"][key] = original = revised + revised = None + phase = "extract" + future = pool.submit(self._repair_extract, key, original, decision, baseline) + pending[future] = (key, original, decision, attempt, repair_base, phase, revised) + if not pending: + break + future = next(as_completed(pending)) + key, original, decision, attempt, repair_base, phase, revised = pending.pop(future) + try: + result = future.result() + if phase == "extract": + self.state.setdefault("repair_drafts", {})[key] = { + "base_sha256": _hash(original), "attempt": attempt, "candidate": deepcopy(result)} + self.save() + ready.appendleft(key) # Judge the durable draft before another correction. + continue + self.state["candidates"][key] = revised self.state.setdefault("repairs", {})[key] = {"base_sha256": repair_base, "candidate": deepcopy(revised)} - decision = {**result, "attempts": attempt} + decision = {**result[key], "attempts": attempt} except BudgetExhausted: - return False + # Stop new dispatch but drain and checkpoint all successes. + stopped = True + continue except ModelUnavailable as exc: - decision = {"supported": "unsure", "relation": "unknown", "related_id": "", "reason": str(exc), - "attempts": decision.get("attempts", 1), "repair_blocked": True} - self.state["reviews"][key] = decision - self.save() # Keep a successful repair draft for an explicit channel retry. - break + self.state["reviews"][key] = {"supported": "unsure", "relation": "unknown", "related_id": "", "reason": str(exc), + "attempts": decision.get("attempts", 1), "repair_blocked": True} + self.save() # Preserve a successful draft for explicit channel retry. + continue except ValueError as exc: decision = {"supported": "unsure", "relation": "unknown", "related_id": "", "reason": str(exc), "attempts": attempt} self.state.get("repair_drafts", {}).pop(key, None) self.state["reviews"][key] = decision self.save() - return True + if eligible(key): + ready.append(key) + return not stopped def _implemented_approval(self, row, judgment): from .feature_discovery_index import validate_evidence @@ -693,7 +725,8 @@ def catalog(self, repo_dir, *, require_boundary=False): result.update(status="unknown", reason="related feature is not accepted in the frozen catalog") continue feature = by_id[target] - for field, additions in (("source_globs", source), ("entry_points", source), ("docs", docs)): + # Relation evidence does not certify an additional entry point. + for field, additions in (("source_globs", source), ("docs", docs)): feature[field] = list(dict.fromkeys(feature.get(field, []) + additions)) return list(by_id.values()), outcomes diff --git a/test/test_kb_feature_discovery.py b/test/test_kb_feature_discovery.py index 96309b60..4c15051e 100644 --- a/test/test_kb_feature_discovery.py +++ b/test/test_kb_feature_discovery.py @@ -143,17 +143,19 @@ def call(role, system, prompt, validate): assert not features and outcomes[0]['status'] == 'unknown' -def test_seed_id_and_owner_are_preserved_during_supplement(tmp_path): +@pytest.mark.parametrize('relation', ['implementation_supplement', 'alias', 'subcapability', 'shared_component']) +def test_seed_id_owner_and_entry_points_are_preserved_during_relation_merge(tmp_path, relation): idx = index(tmp_path, docs=False) seed = {'id': 'session', 'title': 'Existing session', 'owner': 'original', 'source_globs': ['src/original.py'], 'entry_points': ['src/original.py'], 'docs': [], 'page': 'repos/demo/components/original/feature-session.md'} row = candidate(); row['generator_receipts'] = [{'trace_id': 'native-generator'}] - state = {'candidates': {'session': row}, 'reviews': {'session': {'supported': 'yes', 'relation': 'implementation_supplement', + state = {'candidates': {'session': row}, 'reviews': {'session': {'supported': 'yes', 'relation': relation, 'related_id': 'session', 'reason': 'Same lifecycle', 'candidate_sha256': _hash(row), 'judge_receipt': {'trace_id': 'native-judge'}}}} engine = DiscoveryEngine(idx, seeds=[seed], owners=[], state=state, call=None, save=lambda: None) features, _ = engine.catalog('repos/demo') assert features[0]['owner'] == 'original' and features[0]['title'] == 'Existing session' assert features[0]['source_globs'] == ['src/original.py', 'src/session.py'] + assert features[0]['entry_points'] == ['src/original.py'] assert seed['source_globs'] == ['src/original.py'] @@ -393,3 +395,192 @@ def call(role,system,prompt,validate): resumed=DiscoveryEngine(idx,seeds=[],owners=[],state=state,call=call,save=lambda:None) assert resumed.scan() and resumed.review() assert calls.count('generator')==2 and len(resumed.catalog('repos/demo')[0])==1 + + +def rejected_repair_state(count): + rows = [candidate(f'capability-{n}', title=f'Capability {n}') for n in range(count)] + for row in rows: + row.update(generator_receipts=[{'trace_id': 'original'}], origin_rounds=['source']) + return {'candidates': {r['id']: r for r in rows}, 'reviews': { + r['id']: {'supported': 'no', 'relation': 'unknown', 'related_id': '', + 'reason': 'Needs a narrower claim.', 'attempts': 1} for r in rows}} + + +def test_repairs_share_thirteen_slots_and_checkpoint_before_judge(tmp_path): + import threading + import time + idx = index(tmp_path, docs=False) + state = rejected_repair_state(13) + coordinator = threading.get_ident() + barrier = threading.Barrier(13) + lock = threading.Lock() + active = peak = 0 + durable = {} + def save(): + assert threading.get_ident() == coordinator + with lock: + durable.clear() + durable.update(deepcopy(state)) + def call(role, system, prompt, validate): + nonlocal active, peak + payload = json.loads(prompt.split('\n', 1)[1].rsplit('\n', 1)[0]) + with lock: + active += 1 + peak = max(peak, active) + try: + if role == 'generator': + barrier.wait(timeout=5) + row = deepcopy(payload['candidate']) + row['description'] = 'Only return the supplied identity.' + return reply(role, {'candidates': [row]}) + row = payload['candidates'][0] + with lock: + assert _hash(durable['repair_drafts'][row['id']]['candidate']) == _hash(row) + time.sleep(.01) + return reply(role, approve([row])) + finally: + with lock: + active -= 1 + engine = DiscoveryEngine(idx, seeds=[], owners=[], state=state, call=call, save=save, concurrency=13) + assert engine.review() + assert peak == 13 + assert all(r['supported'] == 'yes' and r['attempts'] == 2 for r in state['reviews'].values()) + assert not state['repair_drafts'] + + +def test_parallel_repairs_keep_three_corrections_and_deterministic_content(tmp_path): + from collections import Counter + idx = index(tmp_path, docs=False) + results = [] + for concurrency in (1, 3): + state = rejected_repair_state(7) + calls = Counter() + def call(role, system, prompt, validate): + payload = json.loads(prompt.split('\n', 1)[1].rsplit('\n', 1)[0]) + row = deepcopy(payload['candidate'] if role == 'generator' else payload['candidates'][0]) + calls[(role, row['id'])] += 1 + if role == 'generator': + row['description'] += ' Corrected.' + return reply(role, {'candidates': [row]}) + return reply(role, approve([row], supported='no')) + engine = DiscoveryEngine(idx, seeds=[], owners=[], state=state, call=call, save=lambda: None, + concurrency=concurrency) + assert engine.review() + assert all(r['attempts'] == 4 for r in state['reviews'].values()) + assert all(calls[(role, key)] == 3 for role in ('generator', 'judge') for key in state['candidates']) + previous_calls = calls.copy() + assert engine.review() and calls == previous_calls + results.append(state) + assert results[0] == results[1] + + +def test_parallel_repair_budget_stop_drains_and_resumes_durable_drafts(tmp_path, monkeypatch): + import threading + from collections import Counter + from concurrent.futures import wait + import infermatrix_copilot.kb_service.init_feature_discovery as discovery + idx = index(tmp_path, docs=False) + state = rejected_repair_state(4) + barrier = threading.Barrier(3) + save_seen = threading.Event() + calls = Counter() + def save(): + if any(r['supported'] == 'yes' for r in state['reviews'].values()): + save_seen.set() + def budget_first(futures): + wait(futures) + return iter(sorted(futures, key=lambda f: not isinstance(f.exception(), BudgetExhausted))) + # Deterministically observe the budget failure before in-flight successes. + monkeypatch.setattr(discovery, 'as_completed', budget_first) + # Three already-durable drafts start independent reviews. One budget stop + # must preserve its draft, drain the other approvals and never dispatch #4. + for key, row in list(state['candidates'].items())[:3]: + revised = deepcopy(row); revised['description'] += ' Corrected.' + state.setdefault('repair_drafts', {})[key] = {'base_sha256': _hash(row), 'attempt': 2, 'candidate': revised} + def call(role, system, prompt, validate): + payload = json.loads(prompt.split('\n', 1)[1].rsplit('\n', 1)[0]) + row = payload['candidates'][0] + calls[(role, row['id'])] += 1 + barrier.wait(timeout=5) + if row['id'] == 'capability-0': + raise BudgetExhausted('stop new dispatch') + return reply(role, approve([row])) + engine = DiscoveryEngine(idx, seeds=[], owners=[], state=state, call=call, save=save, concurrency=3) + assert not engine.review() + assert save_seen.is_set() + assert state['reviews']['capability-1']['supported'] == 'yes' + assert state['reviews']['capability-2']['supported'] == 'yes' + assert state['repair_drafts']['capability-0']['attempt'] == 2 + assert state['reviews']['capability-0']['attempts'] == 1 + assert not any(key == 'capability-3' for role, key in calls) + def resumed_call(role, system, prompt, validate): + payload = json.loads(prompt.split('\n', 1)[1].rsplit('\n', 1)[0]) + row = payload['candidate'] if role == 'generator' else payload['candidates'][0] + calls[(role, row['id'])] += 1 + return reply(role, {'candidates': [row]} if role == 'generator' else approve([row])) + resumed = DiscoveryEngine(idx, seeds=[], owners=[], state=json.loads(json.dumps(state)), call=resumed_call, + save=lambda: None, concurrency=3) + assert resumed.review() + assert calls[('generator', 'capability-0')] == 0 + assert calls[('judge', 'capability-1')] == calls[('judge', 'capability-2')] == 1 + assert all(r['supported'] == 'yes' for r in resumed.state['reviews'].values()) + + +def test_parallel_extraction_budget_stop_saves_other_drafts_without_judging(tmp_path, monkeypatch): + import threading + from concurrent.futures import wait + import infermatrix_copilot.kb_service.init_feature_discovery as discovery + idx = index(tmp_path, docs=False) + state = rejected_repair_state(4) + barrier = threading.Barrier(3) + called = [] + def budget_first(futures): + wait(futures) + return iter(sorted(futures, key=lambda f: not isinstance(f.exception(), BudgetExhausted))) + monkeypatch.setattr(discovery, 'as_completed', budget_first) + def call(role, system, prompt, validate): + payload = json.loads(prompt.split('\n', 1)[1].rsplit('\n', 1)[0]) + assert role == 'generator' + row = payload['candidate']; called.append(row['id']) + barrier.wait(timeout=5) + if row['id'] == 'capability-0': + raise BudgetExhausted('stop new dispatch') + row['description'] += ' Corrected.' + return reply(role, {'candidates': [row]}) + engine = DiscoveryEngine(idx, seeds=[], owners=[], state=state, call=call, save=lambda: None, concurrency=3) + assert not engine.review() + assert sorted(called) == ['capability-0', 'capability-1', 'capability-2'] + assert set(state['repair_drafts']) == {'capability-1', 'capability-2'} + assert all(r['attempts'] == 1 for r in state['reviews'].values()) + assert all(d['attempt'] == 2 for d in state['repair_drafts'].values()) + + +@pytest.mark.parametrize('failure_phase', ['generator', 'judge']) +def test_parallel_repair_single_channel_or_content_failure_stays_local(tmp_path, failure_phase): + from collections import Counter + idx = index(tmp_path) + state = rejected_repair_state(5) + state['candidates']['capability-3']['evidence'] = [{'path': 'docs/session.md', 'start': 1, 'end': 2}] + state['reviews']['capability-4']['attempts'] = 4 + calls = Counter() + def call(role, system, prompt, validate): + payload = json.loads(prompt.split('\n', 1)[1].rsplit('\n', 1)[0]) + row = payload['candidate'] if role == 'generator' else payload['candidates'][0] + calls[(role, row['id'])] += 1 + if row['id'] == 'capability-0' and role == failure_phase: + raise ModelUnavailable('one native channel failed') + if role == 'generator': + row = deepcopy(row) + if row['id'] == 'capability-2': + row['id'] = 'wrong-id' + return reply(role, {'candidates': [row]}) + return reply(role, approve([row])) + engine = DiscoveryEngine(idx, seeds=[], owners=[], state=state, call=call, save=lambda: None, concurrency=3) + assert engine.review() + assert state['reviews']['capability-0']['repair_blocked'] + assert state['reviews']['capability-0']['attempts'] == 1 + assert ('capability-0' in state.get('repair_drafts', {})) == (failure_phase == 'judge') + assert state['reviews']['capability-1']['supported'] == 'yes' + assert state['reviews']['capability-2']['attempts'] == 4 + assert calls[('generator', 'capability-2')] == 3 + assert not any(key in ('capability-3', 'capability-4') for role, key in calls) diff --git a/test/test_kb_feature_discovery_boundary.py b/test/test_kb_feature_discovery_boundary.py index b48ae087..e664f2d7 100644 --- a/test/test_kb_feature_discovery_boundary.py +++ b/test/test_kb_feature_discovery_boundary.py @@ -5,6 +5,7 @@ import pytest from infermatrix_copilot.kb_service.init_budget import BudgetExhausted +from infermatrix_copilot.kb_service.feature_discovery_index import build_discovery_index from infermatrix_copilot.kb_service.init_feature_discovery import ( DiscoveryEngine, SYSTEM_REVIEW, _hash, _boundary_ready, ) @@ -144,14 +145,27 @@ def call(role, system, prompt, validate): assert all(r['status'] == 'unknown' and 'duplicate' in r['reason'] for r in outcomes) -def test_aliases_can_resolve_to_an_independently_accepted_new_parent(tmp_path): +@pytest.mark.parametrize('relation', ['implementation_supplement', 'alias', 'subcapability', 'shared_component']) +def test_aliases_can_resolve_to_an_independently_accepted_new_parent(tmp_path, relation): def call(role, system, prompt, validate): - mapping = {'parent': ('new', ''), 'alias-one': ('alias', 'parent'), 'alias-two': ('alias', 'alias-one')} + mapping = {'parent': ('new', ''), 'alias-one': (relation, 'parent'), 'alias-two': ('alias', 'alias-one')} _, data = decisions(prompt, lambda r: mapping[r['id']]); return reply(role, data) e = engine(tmp_path, [candidate(key=k, title=k) for k in ('parent', 'alias-one', 'alias-two')], call) + for key in ('alias-one', 'alias-two'): + path = f'src/{key}.py' + (tmp_path / path).write_text('def helper(identity):\n return identity\n') + row = e.state['candidates'][key] + row['evidence'] = [{'path': path, 'start': 1, 'end': 2}, + {'path': 'docs/session.md', 'start': 1, 'end': 2}] + e.state['reviews'][key] = approved(row) + e.index = build_discovery_index(tmp_path, pin=PIN, scope={'roots': ['src/'], 'exclude': []}, + doc_globs=['docs/*.md']) assert e.audit_catalog_boundaries() features, outcomes = e.catalog('repos/library', require_boundary=True) assert [f['id'] for f in features] == ['parent'] + assert features[0]['entry_points'] == ['src/session.py'] + assert set(features[0]['source_globs']) == {'src/session.py', 'src/alias-one.py', 'src/alias-two.py'} + assert features[0]['docs'] == ['docs/session.md'] assert all(r['status'] == 'accepted' for r in outcomes)