From 33fa2ed28c0186edac65b29f259263e096383150 Mon Sep 17 00:00:00 2001 From: mikamikasuki Date: Mon, 12 Oct 2026 06:56:26 -0700 Subject: [PATCH] Preserve owner resume and consistent execution recovery --- research/agents/processes.py | 46 ++++++- research/agents/research_path.py | 12 ++ research/agents/runtime.py | 3 +- scripts/backup.py | 6 +- services/interventions/lifecycle.py | 15 +++ services/worker/main.py | 5 + tests/test_codex_request_provenance.py | 14 +- tests/test_execution_reconciliation.py | 36 +++++ tests/test_figure_raster_finish.py | 2 +- tests/test_intervention_restore.py | 21 ++- tests/test_manuscript_visual_contract.py | 5 +- tests/test_owner_budget_resume.py | 164 +++++++++++++++++++++++ tests/test_process_record_snapshot.py | 159 ++++++++++++++++++++++ tests/test_research_path_recovery.py | 16 +++ 14 files changed, 484 insertions(+), 20 deletions(-) create mode 100644 tests/test_owner_budget_resume.py create mode 100644 tests/test_process_record_snapshot.py diff --git a/research/agents/processes.py b/research/agents/processes.py index 57f226ac..5f67be7f 100644 --- a/research/agents/processes.py +++ b/research/agents/processes.py @@ -219,6 +219,45 @@ def read_json(path, default=None): return {} if default is None else default +def _record_versions(folder): + versions = {} + for name in ('request', 'state', 'identity', 'completion'): + try: + stat = (folder / (name + '.json')).stat() + versions[name] = (stat.st_dev, stat.st_ino, stat.st_size, + stat.st_mtime_ns, stat.st_ctime_ns) + except FileNotFoundError: + versions[name] = None + return versions + + +def _read_record_snapshot(folder): + """Observe one stable publication without relaxing original bindings. + + The host atomically replaces individual files, so sequential JSON reads + can straddle its launch or completion publication. File versions bracket + all reads, including absent files; only a changed observation is retried. + """ + for attempt in range(3): + before = _record_versions(folder) + error = None + try: + records = (read_json(folder / 'request.json'), + read_json(folder / 'state.json', {'status': 'starting'}), + read_json(folder / 'identity.json'), + read_json(folder / 'completion.json')) + except (OSError, ValueError, UnicodeError) as exc: + error = exc + after = _record_versions(folder) + if before == after: + if error is not None: + raise error + return records, before['state'] is not None + if attempt < 2: + time.sleep(.005) + raise ValueError('Concurrent executor record publication did not settle after three observations') + + def is_alive(pid, created): if not pid or created is None: return False @@ -314,10 +353,7 @@ def start(self, command, *, process_id=None, cwd='.', env=None, timeout=None, ex def inspect(self, process_id): folder = self.folder(process_id) try: - request = read_json(folder / 'request.json') - state = read_json(folder / 'state.json', {'status': 'starting'}) - identity = read_json(folder / 'identity.json') - receipt = read_json(folder / 'completion.json') + (request, state, identity, receipt), state_present = _read_record_snapshot(folder) except (OSError,ValueError,UnicodeError) as exc: return {'process_id':process_id,'status':RECONCILIATION, 'error':'Executor record is unreadable; reconcile original launch records: '+str(exc)} @@ -328,7 +364,7 @@ def inspect(self, process_id): return {**request,**state,'status':RECONCILIATION, 'error':'Legacy or changed executor record requires reconciliation of its original configuration and observations'} if any(record and not binding_matches(request,record) for record in - (state if (folder/'state.json').exists() else {},identity,receipt)): + (state if state_present else {},identity,receipt)): return {**request,**state,'status':RECONCILIATION,'error':'Executor identity/state/receipt work records disagree'} for reference in (request.get('source_references') or {}).values(): try: valid=reference.get('snapshot_metadata')==file_metadata(reference['snapshot_path']) diff --git a/research/agents/research_path.py b/research/agents/research_path.py index 30b616c1..767eda6a 100644 --- a/research/agents/research_path.py +++ b/research/agents/research_path.py @@ -148,6 +148,18 @@ def unchanged_completed_work(node, nodes, edges, completed, runs): if not matching: return False prior=[run for run in runs if run.get('node_id') in matching and run.get('status')=='completed'] if not prior: return False # A graph completion label alone is not an executed result. + # Independent work declared before another task's first execution still + # owns its first execution, even with equal methods. It is not a new clone + # created after seeing a result. Reopening an executed node cannot use this + # exemption to repeat the same work indefinitely. + if not any(run.get('node_id')==node['id'] and run.get('status')=='completed' for run in runs): + from datetime import datetime + try: + declared=datetime.fromisoformat(node['created_at']) + first_execution=min(datetime.fromisoformat(run['created_at']) for run in prior) + if declared <= first_execution:return False + except (KeyError,TypeError,ValueError): + pass # Legacy records still use the available direct evidence below. since=max(run.get('created_at','') for run in prior) new_ids=(node.get('config') or {}).get('new_evidence_run_ids') or [] return not any(run['id'] in new_ids and run.get('status')=='completed' and run.get('created_at','')>since diff --git a/research/agents/runtime.py b/research/agents/runtime.py index 665d57f1..a369adbb 100644 --- a/research/agents/runtime.py +++ b/research/agents/runtime.py @@ -1041,7 +1041,8 @@ def action_result(content, action): action_id = state['wait_for']['action_id'] with Session() as session: receipt = session.get(ToolExecution, action_id) - submitted = deepcopy(receipt.result) if receipt and receipt.run_id == run_id and receipt.status == 'completed' else None + submitted = deepcopy(receipt.result) if (receipt and receipt.run_id == run_id and + receipt.status == 'completed' and receipt.result.get('status') == 'submission_settled') else None if submitted is None: save() raise AgentYield(status='waiting_input', wait_for=state['wait_for'], diff --git a/scripts/backup.py b/scripts/backup.py index 5ded96ff..ef36b967 100644 --- a/scripts/backup.py +++ b/scripts/backup.py @@ -212,6 +212,7 @@ def _drop_postgres(parameters, name): def quarantine_execution(database_url): """Restored process IDs/remote handles must never control old live work.""" from sqlalchemy import inspect, text + from services.interventions.history import historical_resource engine = _engine(database_url) try: with engine.begin() as connection: @@ -226,10 +227,7 @@ def save_object(table,column,ident,value): connection.execute(text(f'UPDATE {table} SET {column}={assignment} WHERE id=:id'),{'value':encoded,'id':ident}) if 'task_runs' in tables and 'resource' in {c['name'] for c in inspect(connection).get_columns('task_runs')}: for ident,resource in connection.execute(text('SELECT id, resource FROM task_runs')).all(): - resource=read_object(resource) - if resource.get('pending_intervention'): - resource['historical_pending_intervention']=resource.pop('pending_intervention') - resource.pop('wait_for',None) + resource=historical_resource(read_object(resource)) save_object('task_runs','resource',ident,resource) if 'intervention_effects' in tables: for ident,status,observation in connection.execute(text('SELECT id,status,observation FROM intervention_effects')).all(): diff --git a/services/interventions/lifecycle.py b/services/interventions/lifecycle.py index 37c01461..19187520 100644 --- a/services/interventions/lifecycle.py +++ b/services/interventions/lifecycle.py @@ -39,6 +39,8 @@ def prepare_resume(s, project, r, body): other_runs=s.scalars(select(TaskRun).where(TaskRun.project_id==r.project_id,TaskRun.id!=r.id)) other_elapsed=sum(_elapsed_seconds_at(item) for item in other_runs) elapsed=_elapsed_seconds_at(r) + if requested_timeout is not None and elapsed >= requested_timeout: + error('TIME_BUDGET_EXHAUSTED','The requested task time budget is exhausted',409) time_budget=dict((r.resource or {}).get('time_budget') or {}) time_budget['requested_task_timeout_seconds']=requested_timeout r.resource={**(r.resource or {}),'elapsed_seconds':elapsed,'time_budget':time_budget} @@ -58,6 +60,19 @@ def prepare_resume(s, project, r, body): 'project_budget_seconds':project_budget.get('seconds'), 'project_revision':project.revision,'other_run_elapsed_seconds':other_elapsed, 'other_run_reserved_seconds':held_by_other,'run_elapsed_seconds':elapsed} + exhausted = resource.get('hard_budget_exhausted') + if isinstance(exhausted, dict) and exhausted.get('kind') == 'seconds': + # A successful explicit resume grants a newly calculated allowance. + # Retire only the old time gate; receipts, elapsed usage and uncertain + # original jobs remain owned by this same run and its supervisor. + resource.pop('hard_budget_exhausted') + if resource.get('blocked_reason') == 'time_budget_exhausted': + resource.pop('blocked_reason') + time_budget['exhaustion_history'] = [*time_budget.get('exhaustion_history', []), + {**deepcopy(exhausted), 'retired_by_explicit_resume': deepcopy(resume_record)}] + r.resource = resource + emit(s,r.project_id,'run_time_budget_reopened',{'run_id':r.id, + 'previous_exhaustion':exhausted,**resume_record}) time_budget.update(project_budget_seconds_at_resume=project_budget.get('seconds'), project_revision_at_resume=project.revision, resume_history=[*time_budget.get('resume_history',[]),resume_record]) diff --git a/services/worker/main.py b/services/worker/main.py index 1e995176..722d76a9 100644 --- a/services/worker/main.py +++ b/services/worker/main.py @@ -195,6 +195,11 @@ def _reconcile_execution_hold(run,output): resource.pop('execution_reconciliation_compute_hold',None) resource['last_reconciled_jobs']=resource.pop('reconciliation_jobs',[]) resource.pop('reconciliation_job',None) + if resource.get('reconciliation_cancel_requested') or resource.get('reconciliation_cancel_error'): + resource['last_reconciliation_cancellation'] = { + 'jobs':resource['last_reconciled_jobs'], + 'requested':resource.pop('reconciliation_cancel_requested',False), + 'error':resource.pop('reconciliation_cancel_error',None),'confirmed_at':now()} if resource.get('hard_budget_exhausted'): resource.update(blocked_reason='time_budget_exhausted') run.status='budget_exhausted';run.exit_code=124 diff --git a/tests/test_codex_request_provenance.py b/tests/test_codex_request_provenance.py index 7c927fdf..56c4385c 100644 --- a/tests/test_codex_request_provenance.py +++ b/tests/test_codex_request_provenance.py @@ -1,5 +1,4 @@ -"""Verify the recorded request digest against the actual inference child input.""" -import hashlib +"""Compare the original recorded request with the actual inference child input.""" import json from types import SimpleNamespace from pathlib import Path @@ -9,15 +8,16 @@ from research.agents.provider import ModelClient -def test_codex_receipt_hashes_actual_cli_prompt(monkeypatch): +def test_codex_receipt_preserves_actual_cli_prompt_and_effective_controls(monkeypatch): from research.agents import codex_transport + from research.agents.policy import RESEARCH_POLICY_MARKER provider = {'id':'test-only', 'kind':'codex_cli', 'base_url':'http://127.0.0.1', 'model':'gpt-6-luna', 'config':{'reasoning_effort':'xhigh'}} client = ModelClient(provider, request_guard=lambda event: None) messages = [{'role':'system','content':'Current controls'}, {'role':'user','content':'Owner correction 中文'}, {'role':'assistant','content':'Previous public action'}] path, payload = client.build_request(messages) - recorded = hashlib.sha256(json.dumps(payload,ensure_ascii=False,sort_keys=True,separators=(',', ':')).encode()).hexdigest() + recorded = json.dumps(payload,ensure_ascii=False,sort_keys=True,separators=(',', ':')).encode() captured = [] def child(argv, **kwargs): captured.append(kwargs['input']) @@ -28,8 +28,10 @@ def child(argv, **kwargs): monkeypatch.setattr(codex_transport.subprocess, 'run', child) client.complete(messages) assert path == 'codex_cli' - assert hashlib.sha256(captured[0].encode()).hexdigest() == recorded - assert json.loads(captured[0])['messages'] == messages + assert captured[0].encode() == recorded + effective = json.loads(captured[0])['messages'] + assert RESEARCH_POLICY_MARKER in effective[0]['content'] + assert effective[1:] == messages def test_codex_action_schema_follows_effective_permissions(monkeypatch): diff --git a/tests/test_execution_reconciliation.py b/tests/test_execution_reconciliation.py index 3f51a749..e3244973 100644 --- a/tests/test_execution_reconciliation.py +++ b/tests/test_execution_reconciliation.py @@ -64,6 +64,42 @@ def failed_cancel(self): completed(manager,job['process_id']) +def test_confirmed_stop_does_not_disable_later_owned_job_timeout(tmp_path): + workspace=tmp_path/'workspace';workspace.mkdir() + manager=ManagedProcesses(workspace) + run=SimpleNamespace(status='waiting_input',pid=None,process_created=None, + config={'timeout':.35},resource={'elapsed_seconds':0},error=None,exit_code=None) + jobs=[] + try: + for index,limit in enumerate((.35,1.5)): + # The explicit resume API validates the refreshed allowance in + # test_owner_budget_resume. This actual executor regression proves + # that each later owned computation still has its own stop attempt. + run.config={'timeout':limit};run.status='waiting_input' + run.resource.pop('hard_budget_exhausted',None) + job=manager.start([sys.executable,'-c','import time;time.sleep(20)'], + process_id='original-'+str(index));jobs.append(job) + (manager.folder(job['process_id'])/'completion.json').write_text('{unreadable') + run.resource=_waiting_outcome_resource(run.resource,{'status':'waiting_input', + 'wait_for':{'type':'execution_reconciliation','process_id':job['process_id'],'workspace':str(workspace)}}) + assert is_alive(job['pid'],job['process_created']) + deadline=time.monotonic()+5 + while run.status!='budget_exhausted' and time.monotonic()=1.5 + assert len(manager.all())==2 + finally: + for job in jobs: + manager.cancel(job['process_id']);completed(manager,job['process_id']) + + @pytest.mark.parametrize('mode',['hard_timeout','receipt_recovered']) def test_actual_worker_reconciliation_hold_enforces_original_hard_timeout(tmp_path,mode): h=Harness(tmp_path) diff --git a/tests/test_figure_raster_finish.py b/tests/test_figure_raster_finish.py index 062213bd..d68035db 100644 --- a/tests/test_figure_raster_finish.py +++ b/tests/test_figure_raster_finish.py @@ -311,7 +311,7 @@ def test_changed_source_or_invalid_reference_fails_before_planner(tmp_path, monk client, generator = RasterModelDouble(), ImageBatchDouble() graph, contract, style = supplied() graph['edges'][0]['label'] = 'Invented edge meaning' - with pytest.raises(ValueError, match='frozen original source graph'): + with pytest.raises(ValueError, match='complete original source graph'): invoke(tmp_path, client, generator, monkeypatch, graph=graph, contract=contract, style=style) assert not client.calls and not generator.calls reference = tmp_path / 'not-a-png.txt' diff --git a/tests/test_intervention_restore.py b/tests/test_intervention_restore.py index f21bea11..168c3926 100644 --- a/tests/test_intervention_restore.py +++ b/tests/test_intervention_restore.py @@ -43,7 +43,7 @@ def test_restored_pending_controls_cannot_target_original_execution(tmp_path,int def restore_scenario(directory): from fastapi.testclient import TestClient from sqlalchemy import create_engine,text - from services.api.db import migrate,Session,settings + from services.api.db import migrate,Session,settings,TaskRun from services.api.main import app from services.interventions import application from services.interventions.controls import decision_gate @@ -63,13 +63,29 @@ def restore_scenario(directory): fork=client.post(f'/api/projects/{pid}/graph/commands',json={'request_id':str(uuid.uuid4()),'expected_revision':2, 'operation':'fork_branch','targets':[node['id']],'params':{'name':'Pending copied workspace'}}).json() assert fork['intervention']['status']=='accepted',fork + from services.interventions.history import EXECUTION_CONTROL_FIELDS + controls={key:{'original_execution_id':run['id']} for key in EXECUTION_CONTROL_FIELDS} + with Session.begin() as session: + source_run=session.get(TaskRun,run['id']) + source_run.resource={**source_run.resource,**controls} archive=directory/'actual-backup.tar.gz';create_backup(settings.data_dir,settings.database_url,archive) restored=restore_backup(archive,directory/'restored',os.environ['RESTORE_TEST_ADMIN']) - engine=create_engine(restored['database_url']) + database_url=restored['database_url'] + if restored.get('database_name'): + from sqlalchemy.engine import make_url + admin_url=make_url(os.environ['RESTORE_TEST_ADMIN']) + database_url=admin_url.set( + drivername='postgresql+psycopg',database=restored['database_name']) + assert make_url(restored['database_url']).password==('***' if admin_url.password is not None else None) + engine=create_engine(database_url) try: with engine.connect() as session: assert session.execute(text('SELECT status FROM task_runs WHERE id=:id'),{'id':run['id']}).scalar_one()=='interrupted' assert session.execute(text('SELECT status FROM action_decisions WHERE id=:id'),{'id':decision['id']}).scalar_one()=='stale' + resource=session.execute(text('SELECT resource FROM task_runs WHERE id=:id'),{'id':run['id']}).scalar_one() + resource=json.loads(resource) if isinstance(resource,str) else resource + assert not EXECUTION_CONTROL_FIELDS.intersection(resource) + assert resource['historical_execution_control']==controls for status,target,lease in session.execute(text('SELECT status,target,lease_until FROM intervention_effects')): assert status=='superseded' and (json.loads(target) if isinstance(target,str) else target)=={} and lease==0 for status,extra in session.execute(text('SELECT status,extra FROM branches')): @@ -80,6 +96,7 @@ def restore_scenario(directory): from services.interventions.models import ActionDecision,Intervention assert session.get(ActionDecision,decision['id']).status=='pending' assert session.get(Intervention,instruction['id']).status=='accepted' + assert all(session.get(TaskRun,run['id']).resource[key]==value for key,value in controls.items()) print(json.dumps({'status':'passed','backend':engine.dialect.name,'actual_restore':True,'pending_effect_authority':False,'original_database_unchanged':True})) finally: engine.dispose() diff --git a/tests/test_manuscript_visual_contract.py b/tests/test_manuscript_visual_contract.py index 326d4d28..51e4bfa2 100644 --- a/tests/test_manuscript_visual_contract.py +++ b/tests/test_manuscript_visual_contract.py @@ -147,7 +147,10 @@ def test_real_candidate_pixels_reach_each_transport(tmp_path,api): message=image_content(client,[paths['png']],'Inspect these actual computed observations.') _,payload=client.build_request([message]) if api=='ollama': - assert payload['messages'][0]['images'] and isinstance(payload['messages'][0]['content'],str) + from research.agents.policy import RESEARCH_POLICY_MARKER + assert RESEARCH_POLICY_MARKER in payload['messages'][0]['content'] + observed=next(item for item in payload['messages'] if item['role']=='user') + assert observed['images']==message['images'] and isinstance(observed['content'],str) else: import base64,io,re from PIL import Image diff --git a/tests/test_owner_budget_resume.py b/tests/test_owner_budget_resume.py new file mode 100644 index 00000000..70b600d4 --- /dev/null +++ b/tests/test_owner_budget_resume.py @@ -0,0 +1,164 @@ +"""Explicit owner admission reopens real allowance without replaying owned work.""" +import json +import sys +import time +import uuid + +import pytest + +from test_worker import wait_until +from test_submission_handoff import SUBMISSION_AGENT, FITTED_CHILD +from tests.test_intervention_worker import intervention_harness + + +def _prepare_parent(h, *, project_seconds, task_timeout=None, job_seconds=20): + h.start_api(); h.start_worker() + project = h.request('POST', '/api/projects', json={ + 'name': 'Owner changes actual compute allowance', 'mode': 'manual', + 'goal': 'Fit and validate after original computation settles', + 'budget': {'seconds': project_seconds}}) + graph = h.request('GET', f"/api/projects/{project['id']}/graph") + branch = graph['branches'][0] + workspace = h.directory / 'data' / 'projects' / project['id'] / branch['workspace'] + workspace.mkdir(parents=True, exist_ok=True) + # The original job is a real detached process. The marker is outside its + # editable workspace and prevents fixture re-entry from launching it again. + injected = ''' from research.execution.process_manager import process_manager + import sys + if not (folder/'original-job-started.txt').exists(): + original_job=process_manager(folder/'workspace',config).start( + [sys.executable,'-c',"import time;from pathlib import Path;Path('original-job-launch.txt').write_text('one launch');time.sleep(JOB_SECONDS);Path('original-job-completed.txt').write_text('completed')"], + process_id='owned-before-submission') + (folder/'original-job-started.txt').write_text(json.dumps(original_job)) + original=runtime.ModelClient +'''.replace('JOB_SECONDS', str(job_seconds)) + (workspace / 'submission_agent.py').write_text(SUBMISSION_AGENT.replace( + ' original=runtime.ModelClient\n', injected)) + (workspace / 'fitted_child.py').write_text(FITTED_CHILD) + source_id = str(uuid.uuid4()) + result = h.request('POST', f"/api/projects/{project['id']}/graph/commands", json={ + 'request_id': str(uuid.uuid4()), 'expected_revision': graph['revision'], 'operation': 'add_node', + 'params': {'id': source_id, 'branch_id': branch['id'], 'type': 'experiment', + 'title': 'Parent with accepted original work', 'instructions': 'Inspect the actual child fit receipt', + 'config': {'kind': 'experiment', 'entrypoint': 'submission_agent:run', 'role': 'Researcher', + 'execution_backend': 'local', 'tools': ['experiment_run', 'finish'], + **({'timeout': task_timeout} if task_timeout is not None else {}), + 'parameters': {'child_delay': .1}}}}) + parent = h.launch(next(node for node in result['graph']['nodes'] if node['id'] == source_id)) + return project, parent + + +def _original_receipt(h, parent): + return json.loads((h.output(parent) / 'workspace' / '.forest-processes' / + 'owned-before-submission' / 'completion.json').read_text()) + + +def _exhausted(h, parent): + value = h.run(parent) + return value if (value['status'] == 'waiting_input' and value['resource'].get('hard_budget_exhausted') + and not value['resource'].get('execution_handoff_compute_hold')) else None + + +def test_owner_budget_extension_resumes_same_intent_after_real_job_cancel(tmp_path, intervention_harness): + h = intervention_harness + try: + project, parent = _prepare_parent(h, project_seconds=3) + before = wait_until(lambda: _exhausted(h, parent), timeout=20) + assert before['resource']['time_budget']['requested_task_timeout_seconds'] is None + assert before['resource']['elapsed_seconds'] >= 3 + assert _original_receipt(h, parent)['status'] == 'cancelled' + original_launch_id = _original_receipt(h, parent)['launch_id'] + pending = before['resource']['pending_submission'] + usage_path = h.output(parent) / 'agent_usage.json' + prior_usage = json.loads(usage_path.read_text()) + h.request('PATCH', f"/api/projects/{project['id']}", json={'budget': {'seconds': 100}}) + resumed = h.request('POST', f"/api/runs/{parent['id']}/resume", json={}) + assert resumed['id'] == parent['id'] + assert not resumed['resource'].get('hard_budget_exhausted') + assert resumed['resource']['elapsed_seconds'] >= before['resource']['elapsed_seconds'] + final = h.terminal(parent, timeout=35) + assert final['status'] == 'completed', final + runs = h.request('GET', f"/api/projects/{project['id']}/runs") + assert len(runs) == 2 + child = next(row for row in runs if row['id'] != parent['id']) + assert child['status'] == 'completed' + receipt = final['resource']['submission_receipts'][pending['action_id']] + assert receipt['run_ids'] == [child['id']] + assert receipt['settled_at'] + measurements = json.loads((h.output(child) / 'real-fit.json').read_text()) + assert measurements['validation_mse'] < .01 + assert measurements == child['metrics'] + assert final['resource']['elapsed_seconds'] >= before['resource']['elapsed_seconds'] + history = final['resource']['time_budget']['exhaustion_history'] + assert history[0]['limit'] == 3 + assert history[0]['retired_by_explicit_resume']['project_budget_seconds'] == 100 + assert _original_receipt(h, parent)['launch_id'] == original_launch_id + assert _original_receipt(h, parent)['status'] == 'cancelled' + assert not (h.output(parent) / 'workspace' / 'original-job-completed.txt').exists() + usage = json.loads(usage_path.read_text()) + assert usage['run_id'] == prior_usage['run_id'] == parent['id'] + assert usage['active_seconds'] >= prior_usage['active_seconds'] + assert len(json.loads((h.output(parent) / 'workspace' / 'agent_session.json').read_text())['transcript']) == 2 + finally: + h.cleanup() + + +@pytest.mark.parametrize('new_project_seconds', [100, None]) +def test_project_growth_cannot_reopen_exhausted_requested_task_limit(tmp_path, intervention_harness, new_project_seconds): + h = intervention_harness + try: + project, parent = _prepare_parent(h, project_seconds=6, task_timeout=3) + before = wait_until(lambda: _exhausted(h, parent), timeout=20) + assert before['resource']['time_budget']['requested_task_timeout_seconds'] == 3 + assert _original_receipt(h, parent)['status'] == 'cancelled' + h.request('PATCH', f"/api/projects/{project['id']}", json={'budget': {'seconds': new_project_seconds}}) + response = h.client.post(f"/api/runs/{parent['id']}/resume", json={}) + assert response.status_code == 409, response.text + assert response.json()['detail']['code'] == 'TIME_BUDGET_EXHAUSTED' + after = h.run(parent) + assert after['resource']['hard_budget_exhausted'] == before['resource']['hard_budget_exhausted'] + assert after['resource']['elapsed_seconds'] >= before['resource']['elapsed_seconds'] + assert len(h.request('GET', f"/api/projects/{project['id']}/runs")) == 1 + finally: + h.cleanup() + + +@pytest.mark.parametrize('uncertain_receipt', [False, True]) +def test_early_manual_resume_waits_for_original_job_and_actual_submission(tmp_path, intervention_harness, uncertain_receipt): + h = intervention_harness + try: + project, parent = _prepare_parent(h, project_seconds=40, job_seconds=4) + waiting = wait_until(lambda: (value if (value := h.run(parent))['status'] == 'waiting_input' + and value['resource'].get('execution_handoff_compute_hold') else None), timeout=20) + accepted = waiting['resource']['pending_submission'] + original = json.loads((h.output(parent) / 'original-job-started.txt').read_text()) + assert original['status'] == 'running' + if uncertain_receipt: + (h.output(parent) / 'workspace' / '.forest-processes' / 'owned-before-submission' / + 'completion.json').write_text('{unreadable original receipt') + waiting = wait_until(lambda: (value if (value := h.run(parent))['resource'].get('blocked_reason') + == 'execution_reconciliation' else None), timeout=10) + resumed = h.request('POST', f"/api/runs/{parent['id']}/resume", json={}) + wait_until(lambda: (value if (value := h.run(parent))['resource'].get('last_intervention', {}).get('id') + == resumed['intervention']['id'] else None), timeout=10) + time.sleep(.5) + current = h.run(parent) + assert current['status'] == 'waiting_input' + assert current['resource']['pending_submission']['action_id'] == accepted['action_id'] + assert len(h.request('GET', f"/api/projects/{project['id']}/runs")) == 1 + assert current['resource']['execution_handoff_compute_hold'] + assert current['resource']['elapsed_seconds'] >= waiting['resource']['elapsed_seconds'] + state = json.loads((h.output(parent) / 'workspace' / 'agent_session.json').read_text()) + assert len(state['transcript']) == 1 + assert state['wait_for'] == {'type': 'research_submission', 'action_id': accepted['action_id']} + final = h.terminal(parent, timeout=30) + assert final['status'] == 'completed', final + assert _original_receipt(h, parent)['launch_id'] == original['launch_id'] + assert _original_receipt(h, parent)['status'] == 'completed' + runs = h.request('GET', f"/api/projects/{project['id']}/runs") + assert len(runs) == 2 + child = next(row for row in runs if row['id'] != parent['id']) + assert child['status'] == 'completed' + assert json.loads((h.output(child) / 'real-fit.json').read_text())['validation_mse'] < .01 + finally: + h.cleanup() diff --git a/tests/test_process_record_snapshot.py b/tests/test_process_record_snapshot.py new file mode 100644 index 00000000..aeb32bfd --- /dev/null +++ b/tests/test_process_record_snapshot.py @@ -0,0 +1,159 @@ +"""Atomic process publications must be observed as a consistent generation. + +These tests execute one actual short CPU command and retain its original +records. Republishing those records at a controlled reader boundary makes +the host's multi-file publication race deterministic without a substitute +execution transport or scientific result. +""" +import json +import sys +from pathlib import Path + +import pytest + +from research.agents import processes +from research.agents.processes import ManagedProcesses, atomic_json +from test_agent_processes import completed + + +@pytest.fixture +def observed_job(tmp_path): + manager = ManagedProcesses(tmp_path) + command = [sys.executable, '-c', + "from pathlib import Path; p=Path('launch_count'); p.write_text('x')"] + process_id = 'publication-observation' + result = completed(manager, manager.start(command, process_id=process_id)['process_id']) + assert result['status'] == 'completed' + folder = manager.folder(process_id) + originals = {name: json.loads((folder / (name + '.json')).read_text()) + for name in ('request', 'state', 'identity', 'completion')} + return manager, process_id, folder, originals + + +@pytest.mark.parametrize('publish_completion', [False, True]) +def test_absent_state_then_atomic_identity_publication_retries_original_snapshot( + observed_job, monkeypatch, publish_completion): + manager, process_id, folder, originals = observed_job + for name in ('state', 'identity', 'completion'): + (folder / (name + '.json')).unlink() + original_read = processes.read_json + state_reads = 0 + + def publish_after_absent_state(path, default=None): + nonlocal state_reads + value = original_read(path, default) + if Path(path) == folder / 'state.json': + state_reads += 1 + if state_reads == 1: + assert value == {'status': 'starting'} + running = {key: value for key, value in originals['state'].items() + if key not in ('exit_code', 'finished_at', 'elapsed_seconds')} + running['status'] = 'running' + atomic_json(folder / 'state.json', running) + atomic_json(folder / 'identity.json', originals['identity']) + if publish_completion: + # The real host commits its original receipt before state. + atomic_json(folder / 'completion.json', originals['completion']) + atomic_json(folder / 'state.json', originals['state']) + return value + + monkeypatch.setattr(processes, 'read_json', publish_after_absent_state) + observed = manager.inspect(process_id) + assert observed['status'] == ('completed' if publish_completion else 'running') + assert observed['launch_id'] == originals['identity']['launch_id'] + assert observed['configuration'] == originals['request']['configuration'] + assert state_reads >= 2 + assert (manager.workspace / 'launch_count').read_text() == 'x' + + +@pytest.mark.parametrize('record_name', ['state', 'identity', 'completion']) +def test_stable_original_binding_mismatch_remains_reconciliation( + observed_job, monkeypatch, record_name): + manager, process_id, folder, originals = observed_job + changed = {**originals[record_name], 'configuration_version': 2} + atomic_json(folder / (record_name + '.json'), changed) + original_read = processes.read_json + state_reads = 0 + + def count_reads(path, default=None): + nonlocal state_reads + if Path(path) == folder / 'state.json': + state_reads += 1 + return original_read(path, default) + + monkeypatch.setattr(processes, 'read_json', count_reads) + observed = manager.inspect(process_id) + assert observed['status'] == 'reconciliation' + assert 'work records disagree' in observed['error'] + assert state_reads == 1 + assert (manager.workspace / 'launch_count').read_text() == 'x' + + +def test_stable_unreadable_original_receipt_remains_reconciliation(observed_job): + manager, process_id, folder, _ = observed_job + (folder / 'completion.json').write_text('{not valid JSON') + observed = manager.inspect(process_id) + assert observed['status'] == 'reconciliation' + assert 'unreadable' in observed['error'] + + +def test_stable_other_launch_identity_remains_reconciliation(observed_job): + manager, process_id, folder, originals = observed_job + atomic_json(folder / 'identity.json', + {**originals['identity'], 'launch_id': 'different-original-launch'}) + observed = manager.inspect(process_id) + assert observed['status'] == 'reconciliation' + assert 'does not belong to the committed launch' in observed['error'] + + +def test_stable_missing_completion_remains_reconciliation(observed_job): + manager, process_id, folder, _ = observed_job + (folder / 'completion.json').unlink() + observed = manager.inspect(process_id) + assert observed['status'] == 'reconciliation' + assert 'no consistent completion receipt' in observed['error'] + + +def test_changed_unreadable_receipt_retries_republished_original(observed_job, monkeypatch): + manager, process_id, folder, originals = observed_job + (folder / 'completion.json').write_text('{not valid JSON') + original_read = processes.read_json + receipt_reads = 0 + + def repair_during_read(path, default=None): + nonlocal receipt_reads + if Path(path) == folder / 'completion.json': + receipt_reads += 1 + if receipt_reads == 1: + try: + return original_read(path, default) + finally: + atomic_json(folder / 'completion.json', originals['completion']) + return original_read(path, default) + + monkeypatch.setattr(processes, 'read_json', repair_during_read) + observed = manager.inspect(process_id) + assert observed['status'] == 'completed' + assert receipt_reads == 2 + assert (manager.workspace / 'launch_count').read_text() == 'x' + + +def test_continuous_atomic_publication_has_bounded_reads(observed_job, monkeypatch): + manager, process_id, folder, originals = observed_job + original_read = processes.read_json + state_reads = 0 + + def republish_every_read(path, default=None): + nonlocal state_reads + value = original_read(path, default) + if Path(path) == folder / 'state.json': + state_reads += 1 + atomic_json(folder / 'state.json', originals['state']) + return value + + monkeypatch.setattr(processes, 'read_json', republish_every_read) + observed = manager.inspect(process_id) + assert observed['status'] == 'reconciliation' + assert 'publication' in observed['error'] + assert 2 <= state_reads <= 4 + assert (manager.workspace / 'launch_count').read_text() == 'x' diff --git a/tests/test_research_path_recovery.py b/tests/test_research_path_recovery.py index 3f05f36b..b8c871bd 100644 --- a/tests/test_research_path_recovery.py +++ b/tests/test_research_path_recovery.py @@ -10,6 +10,22 @@ from research.planning.route_review import route_health +def test_predeclared_independent_task_owns_one_execution_without_enabling_later_clones(): + from research.agents.research_path import unchanged_completed_work + original={'id':'original','branch_id':'main','instructions':'Measure the same method', + 'config':{'kind':'command','command':['python','measure.py']},'inputs':[], + 'created_at':'2026-10-11T01:00:00Z'} + independent={**deepcopy(original),'id':'independent','created_at':'2026-10-11T01:00:01Z'} + clone={**deepcopy(original),'id':'late-clone','created_at':'2026-10-11T03:00:00Z'} + nodes={item['id']:item for item in (original,independent,clone)} + runs=[{'id':'first','node_id':'original','status':'completed','created_at':'2026-10-11T02:00:00Z'}] + assert not unchanged_completed_work(independent,nodes,[],{'original'},runs) + assert unchanged_completed_work(clone,nodes,[],{'original'},runs) + runs.append({'id':'second','node_id':'independent','status':'completed', + 'created_at':'2026-10-11T02:01:00Z'}) + assert unchanged_completed_work(independent,nodes,[],{'original','independent'},runs) + + def fixture(): project={'id':'project','goal':'Measure the prescribed system','revision':8, 'budget':{'seconds':900,'allow_paid':False},