AI Flow

"""Native filesystem events, idle I/O, shared SSE cache and atomic-write recovery.
No Hydrogen tabs or live run files are touched.
"""
import concurrent.futures, json, os, pathlib, socket, subprocess, sys, tempfile, time, urllib.request
binary = str(pathlib.Path(sys.argv[1]).resolve())
with tempfile.TemporaryDirectory(prefix='aiflow-native-watch-') as root:
    run = pathlib.Path(root) / 'run'
    env = dict(os.environ, AIFLOW_DASHBOARD='off')
    subprocess.run([binary, '--run', str(run), '--ai-thread', 'native-watch-test', 'intake', '--prompt-time', '2026-10-11T10:00:00Z', '--engine', 'codex'], env=env, check=True, capture_output=True)
    with socket.socket() as sock:
        sock.bind(('127.0.0.1', 0)); port = sock.getsockname()[1]
    url = f'http://127.0.0.1:{port}/'
    proc = subprocess.Popen([binary, '--run', str(run), 'timeline', 'serve', '--port', str(port)], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
    streams = []
    def get(path):
        return json.load(urllib.request.urlopen(url + path, timeout=3))
    def event(stream):
        while True:
            line = stream.readline().decode()
            assert line, 'SSE closed'
            if line.startswith('data: '): return json.loads(line[6:])
    try:
        for attempt in range(100):
            try: health = get('health'); break
            except OSError: time.sleep(.05)
        else: raise AssertionError('server did not start')
        assert health['stateSource'] == 'native-filesystem-events'
        for _ in range(2):
            stream = urllib.request.urlopen(url + 'events', timeout=20)
            streams.append(stream); assert event(stream)['timeline']['steps']
        baseline = get('health')['stateFileReads']
        # Long enough to cover a heartbeat and 60 old polling iterations per client.
        start = time.monotonic()
        while b'heartbeat' not in streams[0].readline():
            assert time.monotonic() - start < 18
        assert get('health')['stateFileReads'] == baseline, 'idle SSE heartbeat read disk'
        timeline = run / 'timeline'
        (timeline / 'unrelated.txt').write_text('not state')
        time.sleep(.3)
        assert get('health')['stateFileReads'] == baseline, 'unrelated file caused state reads'
        state = get('state')
        assert get('health')['stateFileReads'] == baseline, 'GET /state bypassed cache'
        # In-place external write broadcasts to both subscribers.
        (timeline / 'presentation.json').write_text(json.dumps({'caption':'external write'}))
        for stream in streams: assert event(stream)['presentation']['caption'] == 'external write'
        # Partial writes do not erase the last valid presentation.
        (timeline / 'presentation.json').write_text('{')
        time.sleep(.2)
        assert get('state')['presentation']['caption'] == 'external write'
        # Atomic rename keeps the directory watch alive.
        tmp = timeline / 'external.tmp'
        tmp.write_text(json.dumps({'caption':'atomic rename'})); tmp.replace(timeline / 'presentation.json')
        for stream in streams: assert event(stream)['presentation']['caption'] == 'atomic rename'
        # Workflow file replacement, not only presentation controls.
        state['timeline']['notificationTest'] = 'workflow changed'
        tmp.write_text(json.dumps(state['timeline'])); tmp.replace(timeline / 'timeline.json')
        for stream in streams: assert event(stream)['timeline']['notificationTest'] == 'workflow changed'
        def post(patch):
            request = urllib.request.Request(url+'state', data=json.dumps({'aiThread':'native-watch-test','presentation':patch}).encode(), headers={'Content-Type':'application/json'})
            return json.load(urllib.request.urlopen(request, timeout=3))
        assert post({'caption':'HTTP immediate'})['ok']
        for stream in streams: assert event(stream)['presentation']['caption'] == 'HTTP immediate'
        with concurrent.futures.ThreadPoolExecutor() as pool:
            assert all(r['ok'] for r in pool.map(post, [{'title':'concurrent title'}, {'summary':'concurrent summary'}]))
        current = get('state')['presentation']
        assert current['title'] == 'concurrent title' and current['summary'] == 'concurrent summary'
        replay = urllib.request.urlopen(url+'events', timeout=3); streams.append(replay)
        assert event(replay)['presentation'] == current
        print(json.dumps({'result':'PASS','idleSeconds':round(time.monotonic()-start,2),'idleStateFileReads':0,'subscribers':2,'heartbeatWithoutReads':True,'externalWrite':True,'atomicRename':True,'partialWriteRecovery':True,'workflowUpdate':True,'httpControl':True,'concurrentMerge':True,'reconnectReplay':True}))
    finally:
        for stream in streams: stream.close()
        proc.terminate(); proc.wait(timeout=5)