Input history lives in the sealed vault (input_history table) shared by the REPL and chat TUI; one-time import of the old history file

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

test_tai.py

@@ -6172,23 +6172,94 @@ class TuiChatTests(unittest.TestCase):
         self.assertEqual(hist.newer(), "half typed")
         self.assertIsNone(hist.newer())
 
-    def test_history_cap_and_persistence_shared_file(self):
-        path = os.path.join(tempfile.mkdtemp(), "history")
-        with open(path, "w", encoding="utf-8") as handle:
-            handle.write("from repl\n\nsecond\n")
-        hist = tai.TuiHistory(path)
-        self.assertEqual(hist.entries, ["from repl", "second"])
-        hist.add("multi\nline")
-        again = tai.TuiHistory(path)
-        self.assertEqual(again.entries, ["from repl", "second", "multi line"])
+    def vault(self, passphrase="hist-test-1", profile="t"):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        old = os.environ.get("TAI_HOME")
+        os.environ["TAI_HOME"] = tmp.name
+        self.addCleanup(lambda: os.environ.pop("TAI_HOME", None) if old is None else os.environ.update(TAI_HOME=old))
+
+        class FakeArgs:
+            yes = True
+
+        FakeArgs.profile = profile
+        config = tai.Config(FakeArgs())
+        store = tai.Store(config, tai.Seal(config.home, passphrase))
+        self.addCleanup(store.close)
+        return store, config
+
+    def test_vault_history_roundtrip_dedup_and_cap(self):
+        store, _config = self.vault()
+        self.assertTrue(store.history_add("t", "main", "one"))
+        self.assertFalse(store.history_add("t", "main", "one"))
+        self.assertFalse(store.history_add("t", "main", "   "))
+        store.history_add("t", "coder", "multi\nline")
+        self.assertEqual(store.history_load("t"), ["one", "multi\nline"])
+        for n in range(tai.INPUT_HISTORY_CAP + 25):
+            store.history_add("t", "main", "m%d" % n)
+        loaded = store.history_load("t")
+        self.assertEqual(len(loaded), tai.INPUT_HISTORY_CAP)
+        self.assertEqual(loaded[-1], "m%d" % (tai.INPUT_HISTORY_CAP + 24))
+        count = store.db.execute("SELECT COUNT(*) FROM input_history").fetchone()[0]
+        self.assertEqual(count, tai.INPUT_HISTORY_CAP)
+
+    def test_vault_history_is_sealed_on_disk_and_scoped_per_profile(self):
+        store, _config = self.vault()
+        store.history_add("t", "main", "my secret plan")
+        store.history_add("other", "main", "elsewhere")
+        raw = store.db.execute("SELECT text FROM input_history WHERE profile = 't'").fetchone()[0]
+        self.assertTrue(raw.startswith("tai1$"))
+        self.assertNotIn("secret", raw)
+        self.assertEqual(store.history_load("t"), ["my secret plan"])
+        self.assertEqual(store.history_load("other"), ["elsewhere"])
+        bots = [row[0] for row in store.db.execute("SELECT bot FROM input_history ORDER BY id")]
+        self.assertEqual(bots, ["main", "main"])
+
+    def test_vault_history_unsealed_store_and_sealed_rows_refuse(self):
+        store, config = self.vault(passphrase="")
+        store.history_add("t", "main", "plain")
+        self.assertEqual(store.history_load("t"), ["plain"])
+        store.db.execute("UPDATE input_history SET text = 'tai1$abc'")
+        store.db.commit()
+        with self.assertRaises(tai.SealError):
+            tai.Store(config, tai.Seal(config.home, ""))
+
+    def test_vault_history_imports_old_readline_file_once(self):
+        store, config = self.vault()
+        with open(config.history_path, "w", encoding="utf-8") as handle:
+            handle.write("old one\n\nold two\n")
+        self.assertEqual(store.history_load("t"), ["old one", "old two"])
+        with open(config.history_path, "a", encoding="utf-8") as handle:
+            handle.write("added later\n")
+        self.assertEqual(store.history_load("t"), ["old one", "old two"])
+
+    def test_tui_history_uses_vault_backend_and_survives_restart(self):
+        store, _config = self.vault()
+        agent = type("A", (), {})()
+        agent.store, agent.profile, agent.bot = store, "t", "main"
+        hist = tai.TuiHistory(tai.VaultHistory.for_agent(agent))
+        hist.add("persisted")
+        hist.add("persisted")
+        again = tai.TuiHistory(tai.VaultHistory.for_agent(agent))
+        self.assertEqual(again.entries, ["persisted"])
+        self.assertIsNone(tai.VaultHistory.for_agent(FakeChatAgent()))
+
+    def test_tui_history_counts_backend_failures_and_keeps_memory(self):
+        class Broken:
+            def load(self):
+                raise RuntimeError("down")
+
+            def append(self, text):
+                raise RuntimeError("down")
+
+        hist = tai.TuiHistory(Broken())
+        hist.add("x")
+        self.assertEqual(hist.entries, ["x"])
+        self.assertEqual(hist.write_errors, 1)
         small = tai.TuiHistory(cap=3)
         for n in range(10):
             small.add("m%d" % n)
         self.assertEqual(small.entries, ["m7", "m8", "m9"])
-        blocked = tai.TuiHistory(os.path.join(path, "nope", "history"))
-        blocked.add("x")
-        self.assertEqual(blocked.entries, ["x"])
-        self.assertEqual(blocked.write_errors, 1)
 
     def test_up_down_recall_history_in_composer(self):
         app, _agent, _root, refs = self.make_chat()