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.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647import jsonfrom pathlib import Pathimport sqlite3import subprocessimport tempfileimport unittest
from session_context import SessionContext, git_context
class ContextTests(unittest.TestCase): def test_real_git_worktree_and_orca_repository_join(self): with tempfile.TemporaryDirectory() as temporary: home = Path(temporary) repo, worktree = home / 'project', home / 'topic' subprocess.run(['git', 'init', '-q', str(repo)], check=True) subprocess.run(['git', '-C', str(repo), '-c', 'user.name=Test', '-c', 'user.email=test@example.test', 'commit', '--allow-empty', '-qm', 'initial'], check=True) subprocess.run(['git', '-C', str(repo), 'remote', 'add', 'origin', 'https://user:credential@github.com/owner/project.git'], check=True) subprocess.run(['git', '-C', str(repo), 'worktree', 'add', '-qb', 'feature', str(worktree)], check=True) self.assertEqual(git_context(str(worktree))['repo'], 'github.com/owner/project') self.assertEqual(git_context(str(worktree))['branch'], 'feature') dbpath = home / 'Library/Application Support/Orca/profiles/local-default/profile-state.db' dbpath.parent.mkdir(parents=True) with sqlite3.connect(dbpath) as db: db.execute('CREATE TABLE profile_state_documents (domain TEXT, payload TEXT)') db.executemany('INSERT INTO profile_state_documents VALUES (?,?)', [('repos', json.dumps([{'id': 'repo', 'displayName': 'Project', 'path': str(repo)}])), ('worktreeMeta', json.dumps({'repo::' + str(worktree): {'displayName': 'Fix the thing', 'linkedPR': 42, 'hostId': 'local'}}))]) context = SessionContext(home).get(str(worktree / 'src')) self.assertEqual(context['project'], 'Project') self.assertEqual(context['worktree'], 'Fix the thing') self.assertEqual(context['linked_pr'], 42) self.assertEqual(context['host'], 'Orca') self.assertNotIn('credential', json.dumps(context)) self.assertEqual(git_context(str(home)), {})
def test_pi_records_only_user_messages(self): from attention import human_signal for kind, role in [('message', 'assistant'), ('custom_message', 'user'), ('message', 'toolResult')]: self.assertIsNone(human_signal({'type': kind, 'message': {'role': role, 'content': 'not user input'}}, 'Pi')) self.assertIsNone(human_signal({'type': 'message', 'message': {'role': 'user', 'content': 'You are working inside Orca, a multi-agent IDE. You are a dispatched worker. task: read source'}}, 'Pi')) self.assertEqual(human_signal({'type': 'message', 'message': {'role': 'user', 'content': [{'type': 'text', 'text': 'Fix this'}]}}, 'Pi'), ('message', 'Fix this'))
def test_recorded_workdirs_are_data_not_executed_code(self): from session_context import recorded_workdirs self.assertEqual(recorded_workdirs({'payload': {'type': 'custom_tool_call', 'name': 'exec', 'input': 'await tools.exec_command({"workdir":"/project", "cmd":"pwd"})'}}), ['/project']) self.assertEqual(recorded_workdirs({'payload': {'type': 'custom_tool_call', 'name': 'exec', 'input': 'await tools.exec_command({workdir: getSecretPath()})'}}), []) self.assertEqual(recorded_workdirs({'payload': {'type': 'message', 'name': 'exec', 'input': 'workdir: "/project"'}}), [])