Something went wrong. Try again.
Where attention appears to go across agent threads: collector, web explorer, and native iOS client
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110import copyimport jsonfrom pathlib import Pathimport tempfileimport unittest
from workstreams import build, read
def group(title='Recommendation latency', ids=None, first=100, last=110, threads=None): return {'id': str(first), 'title': title, 'summary': 'Investigate the delay.', 'threads': threads or ['a'], 'event_ids': ids or ['e1'], 'first': first, 'last': last, 'sources': [{'thread': 'a', 'at': last, 'text': 'Find the delay.'}]}
def digest(*groups): return {'at': 120, 'zones': {'America/Chicago': {'days': {'today': {'groups': list(groups)}}}}, 'errors': []}
def snapshot(text='The cause remains unknown.', at=115): return {'at': 120, 'errors': ['Partial coverage.'], 'sessions': [ {'id': 'a', 'provider': 'Codex', 'title': 'Investigate latency', 'updated': at, 'project': 'app', 'repo': 'example/app', 'status': 'completed', 'events': [{'label': 'Assistant', 'at': at, 'text': text}, {'label': 'Turn ended', 'at': at, 'text': ''}]}]}
class Workstreams(unittest.TestCase): def test_cross_day_identity_survives_new_membership_and_rename(self): first = build(snapshot(), digest(group()), now=130) newer = group('Investigate recommendation latency', ['e1', 'e2'], last=120) second = build(snapshot(), digest(newer), first, now=140) self.assertEqual(first['streams'][0]['id'], second['streams'][0]['id']) third = build(snapshot(), digest(group(ids=['e3'], first=140, last=150)), second, now=160) self.assertEqual(first['streams'][0]['id'], third['streams'][0]['id'])
def test_same_repository_and_thread_do_not_merge_different_work(self): data = build(snapshot(at=125), digest(group(), group('Fix color catalog', ['e2'], 120, 123)), now=130) self.assertEqual(len(data['streams']), 2) self.assertEqual(sum(bool(s['reports']) for s in data['streams']), 1) self.assertTrue(next(s for s in data['streams'] if s['title'] == 'Fix color catalog')['reports'])
def test_latency_and_hang_in_same_thread_continue_across_midnight(self): data = build(snapshot(), digest(group('Recommendation API hang'), group('Recommendation API latency', ['next-day'], 120, 130)), now=140) self.assertEqual(len(data['streams']), 1) self.assertEqual(data['streams'][0]['first'], 100)
def test_cross_provider_continuity_requires_named_work_and_repository(self): snap = snapshot() snap['sessions'].append({**snap['sessions'][0], 'id': 'b', 'provider': 'Claude'}) data = build(snap, digest(group(), group(ids=['e2'], first=120, last=130, threads=['b'])), now=140) self.assertEqual(len(data['streams']), 1) self.assertEqual(data['streams'][0]['threads'], ['a', 'b']) snap['sessions'][1]['repo'] = 'example/different' self.assertEqual(len(build(snap, digest(group(), group(ids=['e2'], first=120, last=130, threads=['b'])), now=140)['streams']), 2)
def test_split_groups_do_not_collapse_into_the_previous_identity(self): original = build(snapshot(), digest(group(ids=['e1', 'e2'])), now=130) split = build(snapshot(), digest(group(ids=['e1']), group('Fix color catalog', ['e2'], 120, 123)), original, now=140) self.assertEqual(len(split['streams']), 2)
def test_merged_group_preserves_old_deep_link_as_alias(self): original = build(snapshot(), digest(group(ids=['e1']), group('Fix color catalog', ['e2'], 120, 123)), now=130) merged = build(snapshot(), digest(group('Combined investigation', ['e1', 'e2'], 100, 125)), original, now=140) result = merged['streams'][0] self.assertTrue({s['id'] for s in original['streams']} <= {result['id'], *result['aliases']})
def test_provisional_conversation_link_survives_grouping(self): original = build(snapshot(), digest(), now=130) grouped = build(snapshot(), digest(group()), original, now=140) self.assertIn(original['streams'][0]['id'], grouped['streams'][0]['aliases'])
def test_turn_end_never_completes_goal_and_reports_remain_attributed(self): data = build(snapshot(), digest(group()), now=130) stream = data['streams'][0] self.assertNotIn('status', stream) self.assertEqual(stream['reports'][0]['kind'], 'agent_report') self.assertEqual(stream['unresolved'][0]['text'], 'The cause remains unknown.') self.assertIn('Partial coverage.', data['errors'])
def test_regrouping_does_not_change_previous_input(self): data = build(snapshot(), digest(group()), now=130) before = copy.deepcopy(data) build(snapshot(), digest(group(last=125)), data, now=140) self.assertEqual(data, before)
def test_worker_missing_from_digest_is_visible_separately(self): data = build(snapshot(), digest(), now=130) self.assertTrue(data['streams'][0]['provisional']) self.assertEqual(data['streams'][0]['sessions'][0]['key'], 'Codex:a') self.assertTrue(data['errors'])
def test_old_report_is_not_attributed_to_new_request(self): data = build(snapshot(at=105), digest(group()), now=130) self.assertEqual(data['streams'][0]['reports'], [])
def test_read_envelope_staleness_missing_and_bounds(self): with tempfile.TemporaryDirectory() as directory: p = Path(directory)/'sessions.json' self.assertTrue(read(p)['errors']) p.write_text(json.dumps({'workstreams': build(snapshot(), digest(group()), now=130)})) self.assertIn('Workstream updates are delayed.', read(p)['errors']) p.write_text('{}') self.assertEqual(read(p)['streams'], [])
def test_repeated_snapshot_does_not_change_revision(self): first = build(snapshot(), digest(group()), now=130) second = build(snapshot(), digest(group()), first, now=160) self.assertEqual(first['streams'][0]['revision'], second['streams'][0]['revision'])