Preserve owner resume and consistent execution recovery

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'])

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

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'],

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():

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])

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

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):

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()<deadline:
+                _reconcile_execution_hold(run,tmp_path);time.sleep(.03)
+            assert run.status=='budget_exhausted',run.resource
+            assert completed(manager,job['process_id'])['status']=='cancelled'
+            assert not is_alive(job['pid'],job['process_created'])
+            assert not run.resource.get('reconciliation_cancel_requested')
+            assert not run.resource.get('reconciliation_cancel_error')
+            assert run.resource['last_reconciliation_cancellation']['requested']
+            assert job['process_id'] in run.resource['last_reconciliation_cancellation']['jobs']
+        assert run.resource['elapsed_seconds']>=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)

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'

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()

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