feat(session): bind verification to git state

scripts/test-session-state.py

@@ -17,6 +17,27 @@ SHELLS = [
 
 
 class SessionStateHookTests(unittest.TestCase):
+    def test_git_identity_and_stale_verification(self):
+        with tempfile.TemporaryDirectory() as directory:
+            workspace = Path(directory)
+            env = dict(os.environ, GIT_AUTHOR_NAME='Test', GIT_AUTHOR_EMAIL='test@example.invalid', GIT_COMMITTER_NAME='Test', GIT_COMMITTER_EMAIL='test@example.invalid')
+            subprocess.run(['git', '-C', str(workspace), 'init', '-q'], check=True)
+            (workspace / 'file').write_text('one')
+            subprocess.run(['git', '-C', str(workspace), 'add', 'file'], check=True)
+            subprocess.run(['git', '-C', str(workspace), 'commit', '-qm', 'one'], check=True, env=env)
+            command = [sys.executable, str(ROOT / 'shared/hooks/scripts/session-state.py'), '--workspace', str(workspace)]
+            self.assertEqual(subprocess.run(command + ['--verified', 'check'], capture_output=True).returncode, 0)
+            state = json.loads((workspace / '.ai-session/state.json').read_text())
+            self.assertTrue(state['head'])
+            self.assertEqual(state['verification'][0]['head'], state['head'])
+            (workspace / 'file').write_text('two')
+            subprocess.run(['git', '-C', str(workspace), 'add', 'file'], check=True)
+            subprocess.run(['git', '-C', str(workspace), 'commit', '-qm', 'two'], check=True, env=env)
+            self.assertEqual(subprocess.run(command, capture_output=True).returncode, 0)
+            state = json.loads((workspace / '.ai-session/state.json').read_text())
+            self.assertTrue(state['verificationStale'])
+            self.assertTrue(state['verification'][0]['stale'])
+
     def test_writer_migrates_legacy_state_and_requires_reset_for_new_goal(self):
         with tempfile.TemporaryDirectory() as directory:
             workspace = Path(directory)

shared/hooks/scripts/session-state.py

@@ -4,6 +4,8 @@ import os
 from pathlib import Path
 import sys
 import tempfile
+import subprocess
+from urllib.parse import urlsplit, urlunsplit
 from datetime import datetime, timezone
 
 
@@ -26,6 +28,28 @@ def show_pointer(path):
         print('Task state is available at ' + str(path) + '; validate its workspace and goal before reuse.')
 
 
+def git_identity(workspace):
+    try:
+        def git(*args):
+            return subprocess.check_output(['git', '-C', str(workspace), *args], stderr=subprocess.DEVNULL, text=True).strip()
+        root = Path(git('rev-parse', '--show-toplevel')).resolve()
+        try:
+            branch = git('symbolic-ref', '--quiet', '--short', 'HEAD') or 'HEAD'
+        except subprocess.CalledProcessError:
+            branch = 'HEAD'
+        head = git('rev-parse', 'HEAD')
+        try:
+            repository = git('config', '--get', 'remote.origin.url') or str(root)
+            if '://' in repository:
+                parsed = urlsplit(repository)
+                repository = urlunsplit((parsed.scheme, parsed.hostname or '', parsed.path, '', ''))
+        except subprocess.CalledProcessError:
+            repository = str(root)
+        return {'repository': repository, 'branch': branch, 'head': head}
+    except (OSError, subprocess.SubprocessError):
+        return None
+
+
 def hook_pointer():
     try:
         raw = '' if sys.stdin.isatty() else sys.stdin.buffer.read(1048577).decode('utf-8-sig')
@@ -47,18 +71,25 @@ def update_state(args):
     state = json.loads(path.read_text(encoding='utf-8-sig')) if path.exists() else {}
     if not isinstance(state, dict):
         raise ValueError('Invalid task state')
+    for metadata in ('repository', 'branch', 'head'):
+        if metadata in state and not isinstance(state[metadata], str):
+            raise ValueError('Invalid task state metadata')
+    if 'verificationStale' in state and not isinstance(state['verificationStale'], bool):
+        raise ValueError('Invalid task state metadata')
     workspace = str(path.parent.parent)
     if state.get('workspace', workspace) != workspace:
         raise ValueError('Task state belongs to another workspace')
     for old, new in (('changedFiles', 'changed'), ('pending', 'open'), ('importantFindings', 'decisions')):
         if old in state:
             state.setdefault(new, state.pop(old))
-    for field in ('goal', 'decisions', 'changed', 'verified', 'open', 'risks'):
+    for field in ('goal', 'decisions', 'changed', 'verified', 'open', 'risks', 'verification'):
         if field not in state:
             continue
         value = state[field]
         if field == 'goal':
             valid = isinstance(value, str)
+        elif field == 'verification':
+            valid = isinstance(value, list) and all(isinstance(item, dict) and isinstance(item.get('check'), str) and isinstance(item.get('result'), str) for item in value)
         else:
             valid = isinstance(value, list) and all(isinstance(item, str) for item in value)
         if not valid:
@@ -68,12 +99,28 @@ def update_state(args):
     elif args.goal is not None and state.get('goal') and state['goal'] != args.goal:
         raise ValueError('A different goal requires reset')
     state['workspace'] = workspace
+    previous_identity = {key: state.get(key) for key in ('repository', 'branch', 'head')}
+    identity = git_identity(workspace)
+    if identity:
+        state.update(identity)
+        if path.exists() and any(previous_identity.get(key) not in (None, identity[key]) for key in identity):
+            state['verificationStale'] = True
+            for item in state.get('verification', []):
+                item['stale'] = True
+    else:
+        for key in ('repository', 'branch', 'head'):
+            state.pop(key, None)
     for field in ('goal', 'decisions', 'changed', 'verified', 'open', 'risks'):
         value = getattr(args, field)
         if value is not None:
             state[field] = value
         else:
             state.setdefault(field, '' if field == 'goal' else [])
+    state.setdefault('verification', [])
+    if getattr(args, 'verified', None):
+        state['verification'] = [dict(check=check, head=identity['head'], result='pass') for check in args.verified] if identity else state.get('verification', [])
+        if identity:
+            state['verificationStale'] = False
     state['updatedAt'] = datetime.now(timezone.utc).isoformat()
     path.parent.mkdir(exist_ok=True)
     temporary = None
@@ -99,7 +146,7 @@ def main():
     parser.add_argument('--pointer', action='store_true')
     parser.add_argument('--reset', action='store_true')
     parser.add_argument('--goal')
-    for field in ('decisions', 'changed', 'verified', 'open', 'risks'):
+    for field in ('decisions', 'changed', 'verified', 'open', 'risks', 'verification'):
         parser.add_argument('--' + field, nargs='*')
     args = parser.parse_args()
     if args.hook:
@@ -111,11 +158,14 @@ def main():
                 if not isinstance(values, dict):
                     raise ValueError('Invalid task state update')
                 for field, value in values.items():
-                    if field not in ('workspace', 'goal', 'decisions', 'changed', 'verified', 'open', 'risks'):
+                    if field not in ('workspace', 'goal', 'decisions', 'changed', 'verified', 'open', 'risks', 'verification'):
                         raise ValueError('Invalid task state field')
                     if field in ('workspace', 'goal'):
                         if not isinstance(value, str):
                             raise ValueError('Invalid task state value')
+                    elif field == 'verification':
+                        if not isinstance(value, list) or any(not isinstance(item, dict) or not isinstance(item.get('check'), str) or not isinstance(item.get('result'), str) for item in value):
+                            raise ValueError('Invalid verification list')
                     elif not isinstance(value, list) or any(not isinstance(item, str) for item in value):
                         raise ValueError('Invalid task state list')
                     setattr(args, field, value)