diff --git a/scripts/test-session-state.py b/scripts/test-session-state.py index 0f01b50..05d3805 100644 --- a/scripts/test-session-state.py +++ b/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) diff --git a/shared/hooks/scripts/session-state.py b/shared/hooks/scripts/session-state.py index d23f9e9..e8f32e5 100644 --- a/shared/hooks/scripts/session-state.py +++ b/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)