fix(effects): fence tool, Git and maintenance completions (#82, #84, #85)

scripts/ci_shards.py

@@ -24,7 +24,7 @@ def partition(nodeids, index, count):
 
 def evidence(nodeids, index, count):
     all_nodes = sorted(nodeids)
-    return {'version': 1, 'index': index, 'count': count, 'all': all_nodes,
+    return {'version': 2, 'index': index, 'count': count, 'all': all_nodes,
             'selected': partition(all_nodes, index, count)}
 
 
@@ -48,6 +48,14 @@ def verify(directory, count):
         cases = root.findall('.//testcase')
         if len(cases) != len(selected) or root.findall('.//error') or root.findall('.//failure'):
             raise ValueError('JUnit result is incomplete or failed')
+        observed = []
+        for case in cases:
+            identities = case.findall('./properties/property[@name="msg.nodeid"]')
+            if len(identities) != 1 or identities[0].get('value') is None:
+                raise ValueError('JUnit test identity is missing or ambiguous')
+            observed.append(identities[0].get('value'))
+        if sorted(observed) != sorted(selected):
+            raise ValueError('JUnit test identities do not match the selected shard')
     if not expected or seen != expected:
         raise ValueError('test coverage is incomplete')
     return len(seen)
@@ -66,6 +74,10 @@ def pytest_collection_modifyitems(self, config, items):
             chosen = set(manifest['selected'])
             rejected = [item for item in items if item.nodeid not in chosen]
             items[:] = [item for item in items if item.nodeid in chosen]
+            for item in items:
+                # Preserve the exact pytest identity, including class/parameter
+                # punctuation, instead of reverse-engineering JUnit display names.
+                item.user_properties.append(('msg.nodeid', item.nodeid))
             config.hook.pytest_deselected(items=rejected)
             (directory / f'shard-{index}.json').write_text(
                 json.dumps(manifest, ensure_ascii=False, indent=2) + '\n')

src/msg/extensions/repositories.py

@@ -708,6 +708,7 @@ async def push(ctx,request,tx):
 
 async def execute_push(app,job):
     from msg.workers.effects import current_principal,worker_context,effect_request
+    from msg.workers.leases import current_attempt
     store=NativeGitStore(app)
     app.settings.server.staging_dir.mkdir(parents=True,exist_ok=True)
     with tempfile.TemporaryDirectory(prefix='git-push-',dir=app.settings.server.staging_dir) as temp:
@@ -729,8 +730,9 @@ async def execute_push(app,job):
                 context,request=worker_context(app,job,principal),effect_request(app,job,principal)
                 await check_access(app,context,request,tx,job.arguments['id'],'write')
                 require(tx.setting('runtime_config',{}).get('accept_writes',True),'writes_paused')
-                current=await tx.job(job.id)
-                require(current.state=='running' and current.attempts==job.attempts,'job_lease_lost')
+                current=await current_attempt(app,tx,job)
+                if current is None:
+                    return
                 await store.update_refs(job.arguments['id'],job.arguments['changes'])
                 moved=True
                 resource=await tx.resource(job.arguments['id'])

src/msg/workers/effects.py

@@ -23,6 +23,7 @@
 from msg.plugins.common import check_access, create_resource
 from msg.security.policy import scope_subset
 from msg.security.network import intersect_policy
+from msg.workers.leases import current_attempt
 
 
 @dataclass(frozen=True, slots=True)
@@ -131,28 +132,18 @@ async def _claim(self):
             await tx.save_job(job)
             return job, True
 
-    async def _live_attempt(self, tx, job):
-        """Fence completions even when no other worker has swept expired leases."""
-        current = await tx.job(job.id)
-        if current.state != 'running' or current.attempts != job.attempts:
-            return None
-        if current.lease_until is None or current.lease_until <= self.app.clock():
-            await tx.save_job(replace(current, state='uncertain', lease_until=None))
-            tx.set_setting('job_status:' + job.id, {'code': 'expired_execution_lease'})
-            return None
-        return current
-
-    async def _finish(self, job, state, code):
+    async def _finish(self, job, state, code, *, status=None):
         async with self.app.metadata.transaction(write=True) as tx:
-            current = await self._live_attempt(tx, job)
+            current = await current_attempt(self.app, tx, job)
             if current is None:
                 return
             await tx.save_job(replace(current, state=state, lease_until=None))
-            tx.set_setting('job_status:' + job.id, {'code': code})
+            tx.set_setting('job_status:' + job.id,
+                           {'code': code} if status is None else status)
 
     async def _retry(self, job, retry_code, exhausted_code):
         async with self.app.metadata.transaction(write=True) as tx:
-            current = await self._live_attempt(tx, job)
+            current = await current_attempt(self.app, tx, job)
             if current is None:
                 return
             exhausted = current.attempts >= 8
@@ -223,8 +214,9 @@ async def pieces():
             blob = await self.app.contents.put(pieces(), result.media_type)
             await self.app.contents.pin(blob, 'job:'+job.id)
             async with self.app.metadata.transaction(write=True) as tx:
-                current = await tx.job(job.id)
-                require(current.state == 'running' and current.attempts == job.attempts, 'job_lease_lost')
+                current = await current_attempt(self.app, tx, job)
+                if current is None:
+                    return
                 principal = await current_principal(self.app, job.principal, tx)
                 context, request = worker_context(self.app, job, principal), effect_request(self.app, job, principal)
                 await check_access(self.app, context, request, tx, ref.id, 'tool_use')
@@ -357,9 +349,7 @@ async def run_once(self):
             elif job.kind == 'maintenance':
                 from msg.workers.maintenance import run_maintenance
                 result=await run_maintenance(self.app,job.arguments['action'],principal=job.principal)
-                await self._finish(job,'done','ok')
-                async with self.app.metadata.transaction(write=True) as tx:
-                    tx.set_setting('job_status:'+job.id,result)
+                await self._finish(job,'done','ok',status=result)
             elif job.kind == 'gc.resource':
                 # Purge permission was consumed by the committed tombstone. This
                 # idempotent cleanup cannot change another resource's retention.

src/msg/workers/leases.py

@@ -0,0 +1,19 @@
+"""One transaction-local fence for publishing any durable effect's result."""
+from dataclasses import replace
+
+
+async def current_attempt(app, tx, job):
+    """Return the live attempt, or persist uncertainty without resending work.
+
+    Call within the same write transaction as the protected state/ref update.
+    A stale callback must not overwrite another attempt's status, even when no
+    separate worker has swept the expired lease yet.
+    """
+    current = await tx.job(job.id)
+    if current.state != 'running' or current.attempts != job.attempts:
+        return None
+    if current.lease_until is None or current.lease_until <= app.clock():
+        await tx.save_job(replace(current, state='uncertain', lease_until=None))
+        tx.set_setting('job_status:' + job.id, {'code': 'expired_execution_lease'})
+        return None
+    return current

tests/test_ci_shards.py

@@ -1,6 +1,9 @@
 """The CI gate must reject missing, duplicate, changed or failing evidence."""
 import importlib.util
 import json
+import os
+import subprocess
+import sys
 from pathlib import Path
 import xml.etree.ElementTree as ET
 
@@ -20,7 +23,9 @@ def write_evidence(directory, count=4):
         (directory / f'shard-{index}.json').write_text(json.dumps(record))
         suite = ET.Element('testsuite')
         for node in record['selected']:
-            ET.SubElement(suite, 'testcase', name=node)
+            case = ET.SubElement(suite, 'testcase', name=node)
+            properties = ET.SubElement(case, 'properties')
+            ET.SubElement(properties, 'property', name='msg.nodeid', value=node)
         ET.ElementTree(suite).write(directory / f'tests-{index}.xml')
 
 
@@ -73,3 +78,45 @@ def test_gate_rejects_incomplete_or_inconsistent_evidence(tmp_path, damage):
         tree.write(report)
     with pytest.raises((ValueError, FileNotFoundError)):
         ci.verify(tmp_path, 4)
+
+
+@pytest.mark.parametrize('damage', ['renamed_node', 'duplicate_node', 'missing_identity', 'duplicate_identity'])
+def test_equal_test_counts_do_not_hide_wrong_execution_identities(tmp_path, damage):
+    write_evidence(tmp_path)
+    report = tmp_path / 'tests-1.xml'
+    tree = ET.parse(report)
+    cases = tree.getroot().findall('testcase')
+    if damage == 'renamed_node':
+        cases[0].find('./properties/property').set('value', 'tests/test_other.py::wrong')
+    elif damage == 'duplicate_node':
+        cases[0].find('./properties/property').set('value',
+            cases[1].find('./properties/property').get('value'))
+    elif damage == 'missing_identity':
+        cases[0].remove(cases[0].find('properties'))
+    else:
+        ET.SubElement(cases[0].find('properties'), 'property', name='msg.nodeid', value='extra')
+    tree.write(report)
+    with pytest.raises(ValueError):
+        ci.verify(tmp_path, 4)
+
+
+def test_real_pytest_reports_exact_nodeids_for_parameters_and_classes(tmp_path):
+    tests = tmp_path / 'tests'
+    tests.mkdir()
+    (tests / 'test_sample.py').write_text(
+        "import pytest\n"
+        "@pytest.mark.parametrize('value', ['normal', '::[x]', '中文'])\n"
+        "def test_parameter(value):\n    assert value\n"
+        "class TestNested:\n    def test_child(self):\n        assert True\n")
+    directory = tmp_path / 'reports'
+    completed = subprocess.run(
+        [sys.executable, ci.__file__, 'run', '0', '1', '--directory', str(directory)],
+        cwd=tmp_path, env={**os.environ, 'PYTEST_DISABLE_PLUGIN_AUTOLOAD': '1'},
+        capture_output=True, text=True, timeout=30)
+    assert completed.returncode == 0, completed.stdout + completed.stderr
+    assert ci.verify(directory, 1) == 4
+    manifest = json.loads((directory / 'shard-0.json').read_text())
+    tree = ET.parse(directory / 'tests-0.xml')
+    observed = [case.find('./properties/property[@name="msg.nodeid"]').get('value')
+                for case in tree.getroot().findall('.//testcase')]
+    assert sorted(observed) == manifest['selected']

tests/test_effect_completion_fencing.py

@@ -0,0 +1,160 @@
+"""Every completion uses the same stored lease/attempt fence.
+
+Maintenance doubles below isolate state transitions; the tool/Git tests exercise
+signed Operations, real PostgreSQL and local Git/CAS, without external traffic.
+"""
+from dataclasses import replace
+from datetime import timedelta
+
+import pytest
+
+from msg.core.models import EffectJob, Principal
+from msg.workers.effects import EffectWorker, ToolResult
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('outcome', ['expired', 'missing', 'superseded', 'uncertain', 'live'])
+async def test_maintenance_result_is_atomic_with_the_current_attempt(installed, monkeypatch, outcome):
+    from msg.workers import maintenance
+    app, _ = installed
+    now = app.clock()
+    job = EffectJob(id='job_maintenance_fence', event_id='e_maintenance_fence',
+        kind='maintenance', dedupe_key='maintenance-fence',
+        principal=Principal(actor=None, subject=None, credential_id=None,
+                            method='anonymous', certificates=(), ceiling=()),
+        operation='system.maintenance', arguments={'action': 'rebuild_search'},
+        state='pending', attempts=0, next_attempt_at=now, lease_until=None)
+    async with app.metadata.transaction(write=True) as tx:
+        await tx.enqueue(job)
+    report = {'indexed_resources': 2}
+    observed = []
+
+    async def completed_work(app, action, *, principal):
+        assert action == 'rebuild_search'
+        async with app.metadata.transaction(write=True) as tx:
+            current = await tx.job(job.id)
+            if outcome == 'expired':
+                current = replace(current, lease_until=now)
+            elif outcome == 'missing':
+                current = replace(current, lease_until=None)
+            elif outcome == 'superseded':
+                current = replace(current, attempts=current.attempts + 1)
+            elif outcome == 'uncertain':
+                current = replace(current, state='uncertain', lease_until=None)
+            await tx.save_job(current)
+            tx.set_setting('job_status:' + job.id, {'code': 'existing_decision'})
+            observed.append(current)
+        return report
+
+    monkeypatch.setattr(maintenance, 'run_maintenance', completed_work)
+    assert await EffectWorker(app).run_once()
+    async with app.metadata.transaction(write=False) as tx:
+        current = await tx.job(job.id)
+        status = tx.setting('job_status:' + job.id)
+    if outcome == 'live':
+        assert current == replace(observed[0], state='done', lease_until=None)
+        assert status == report
+    elif outcome in {'expired', 'missing'}:
+        assert current == replace(observed[0], state='uncertain', lease_until=None)
+        assert status == {'code': 'expired_execution_lease'}
+    else:
+        assert current == observed[0]
+        assert status == {'code': 'existing_decision'}
+
+
+async def expire(app, job_id, deadline):
+    async with app.metadata.transaction(write=True) as tx:
+        current = await tx.job(job_id)
+        until = {'exact': app.clock(), 'past': app.clock() - timedelta(seconds=1),
+                 'missing': None}[deadline]
+        await tx.save_job(replace(current, lease_until=until))
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('deadline', ['exact', 'past', 'missing'])
+async def test_real_tool_expiry_does_not_publish_a_file_without_a_sweep(installed, deadline):
+    from test_service import register, call
+    from test_authorization import approve, scoped
+    app, root = installed
+    key, uid, _ = await register(app, 'tool-completion-fence')
+    grant = scoped(app, 'tool.use', 'tool_dns', app.registry.capability('tool.use').operations)
+    certificate = await approve(app, root, uid, key, (grant,))
+    accepted = await call(app, 'tool.run', {'id': '/tools/dns',
+        'arguments': {'name': 'example.org', 'type': 'A'}}, key=key,
+        subject=uid, certs=(certificate.resource_id,))
+    assert accepted.status == 'accepted'
+    job_id = accepted.data['job_id']
+    executions = []
+
+    async def runner(tool, arguments, policies, directory):
+        executions.append(tool.executor_key)
+        await expire(app, job_id, deadline)
+        path = directory / 'result.json'
+        path.write_bytes(b'[]')
+        return ToolResult(path=path, media_type='application/json', metadata={})
+
+    worker = EffectWorker(app, tool_runner=runner)
+    assert await worker.run_once()
+    async with app.metadata.transaction(write=False) as tx:
+        current = await tx.job(job_id)
+        assert current.state == 'uncertain' and current.result is None
+        assert tx.setting('job_status:' + job_id) == {'code': 'expired_execution_lease'}
+        assert tx.one('SELECT COUNT(*) FROM resources WHERE name=?', ('tool-' + job_id,))[0] == 0
+    assert executions == ['dns']
+    assert not await worker.run_once()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('deadline', ['exact', 'past', 'missing'])
+async def test_real_git_expiry_does_not_move_refs_without_a_sweep(installed, tmp_path, monkeypatch, deadline):
+    import subprocess
+    from msg.core.codec import b64, wire
+    from msg.extensions.repositories import NativeGitStore
+    from test_service import register, call
+    app, _ = installed
+    key, uid, _ = await register(app, 'git-completion-fence')
+    repo = await call(app, 'git.create', {'parent': '/@git-completion-fence',
+        'name': 'code.git'}, key=key, subject=uid)
+    assert repo.status == 'ok'
+    work = tmp_path / 'work'
+    work.mkdir()
+
+    def git(*arguments):
+        return subprocess.run(['git', '-C', str(work), *arguments], check=True,
+                              capture_output=True).stdout.decode().strip()
+
+    git('init', '--initial-branch=main')
+    git('config', 'user.name', 'Test')
+    git('config', 'user.email', 'test@example.invalid')
+    (work / 'README.md').write_text('public code\n')
+    git('add', '.')
+    git('commit', '-m', 'first')
+    oid = git('rev-parse', 'HEAD')
+    bundle = tmp_path / 'input.bundle'
+    git('bundle', 'create', str(bundle), 'main')
+    upload = await call(app, 'content.file_put', {'parent': '/@git-completion-fence/files',
+        'name': 'input.bundle', 'data': b64(bundle.read_bytes()),
+        'media_type': 'application/octet-stream'}, key=key, subject=uid)
+    assert upload.status == 'ok'
+    accepted = await call(app, 'git.push', {'id': repo.resources[0].id,
+        'bundle': wire(upload.resources[0]),
+        'changes': [{'ref': 'refs/heads/main', 'old': None, 'new': oid}]},
+        key=key, subject=uid)
+    assert accepted.status == 'accepted'
+    job_id = accepted.data['job_id']
+    import_bundle = NativeGitStore.import_bundle
+
+    async def import_then_expire(store, *args, **kwargs):
+        await import_bundle(store, *args, **kwargs)
+        await expire(app, job_id, deadline)
+
+    monkeypatch.setattr(NativeGitStore, 'import_bundle', import_then_expire)
+    worker = EffectWorker(app)
+    assert await worker.run_once()
+    async with app.metadata.transaction(write=False) as tx:
+        assert (await tx.job(job_id)).state == 'uncertain'
+        assert tx.setting('job_status:' + job_id) == {'code': 'expired_execution_lease'}
+        assert (await tx.resource(repo.resources[0].id)).generation == repo.data['generation']
+    refs = await call(app, 'git.refs', {'id': repo.resources[0].id})
+    assert refs.status == 'ok' and list(refs.data['refs']) == []
+    assert not await worker.run_once()