ask_user tool, prompt-JSON CLI repair loop, tool-loop research archive

- New ask_user core tool: mid-turn user question with option picking,
  refusing cleanly in autonomous mode, for background workers, and
  without a tty; system prompt points at it instead of prose questions.
- Structural loop control for prompt-JSON CLI backends (opencode,
  grok, claude, codex, gemini): shared cli_json_turn repairs a
  missing JSON envelope exactly once, then accepts prose; prompt
  contract hardened against narrate-without-calling.
- research/agentic-tool-loop-design: 74 sources plus 2 conclusions
  backing the loop-control decisions; INDEX updated to 6 subjects.

test_tai.py

@@ -3,7 +3,9 @@ import io
 import json
 import os
 import re
+import shutil
 import sqlite3
+import subprocess
 import sys
 import tempfile
 import time
@@ -100,6 +102,120 @@ class ToolTests(unittest.TestCase):
         self.assertIn("Communicate", tai.BOX_TTS)
 
 
+class AskUserTests(unittest.TestCase):
+    def tools(self, **attrs):
+        app = FakeApp()
+        for key, value in attrs.items():
+            setattr(app, key, value)
+        return tai.Tools(app)
+
+    def tty(self):
+        fake_stdin = io.StringIO("")
+        fake_stdin.isatty = lambda: True
+        return mock.patch.object(sys, "stdin", fake_stdin)
+
+    def test_schema_registered_and_core(self):
+        schemas = {schema["function"]["name"]: schema["function"] for schema in tai.TOOL_SCHEMAS}
+        self.assertIn("ask_user", schemas)
+        self.assertIn("question", schemas["ask_user"]["parameters"]["required"])
+        self.assertIn("ask_user", tai.CORE_TOOLS)
+        self.assertIn("question", tai.TOOL_TAGS["ask_user"])
+        self.assertIn("ask_user", {schema["function"]["name"] for schema in tai.select_tools("hello")})
+
+    def test_system_prompt_points_at_tool(self):
+        self.assertIn("ask_user", tai.DEFAULT_SYSTEM)
+        self.assertIn("ask_user", tai.AUTONOMOUS_NOTE)
+
+    def test_validation(self):
+        tools = self.tools()
+        self.assertIn("non-empty string", tools.dispatch("ask_user", json.dumps({})))
+        self.assertIn("non-empty string", tools.dispatch("ask_user", json.dumps({"question": "  "})))
+        self.assertIn("must be a list", tools.dispatch("ask_user", json.dumps({"question": "q?", "options": "nope"})))
+
+    def test_free_input(self):
+        tools = self.tools()
+        with self.tty(), mock.patch("builtins.input", return_value="do the simple thing") as asked, mock.patch("builtins.print") as shown:
+            result = tools.dispatch("ask_user", json.dumps({"question": "Which approach?"}))
+        self.assertEqual(result, "user answered: do the simple thing")
+        self.assertTrue(tools.app.user_stepped_in)
+        asked.assert_called_once()
+        self.assertIn("free input", asked.call_args.args[0])
+        printed = "\n".join(str(call.args[0]) for call in shown.call_args_list)
+        self.assertIn("Which approach?", printed)
+
+    def test_option_by_number(self):
+        tools = self.tools()
+        with self.tty(), mock.patch("builtins.input", return_value="2"), mock.patch("builtins.print") as shown:
+            result = tools.dispatch("ask_user", json.dumps({"question": "Pick one", "options": ["Ship it", "Staging first", "Abort"]}))
+        self.assertEqual(result, "user chose option 2: Staging first")
+        printed = "\n".join(str(call.args[0]) for call in shown.call_args_list)
+        self.assertIn("1) Ship it", printed)
+        self.assertIn("3) Abort", printed)
+
+    def test_option_by_text_case_insensitive(self):
+        tools = self.tools()
+        with self.tty(), mock.patch("builtins.input", return_value="staging FIRST"), mock.patch("builtins.print"):
+            result = tools.dispatch("ask_user", json.dumps({"question": "Pick one", "options": ["Ship it", "Staging first"]}))
+        self.assertEqual(result, "user chose option 2: Staging first")
+
+    def test_out_of_range_number_is_free_text(self):
+        tools = self.tools()
+        with self.tty(), mock.patch("builtins.input", return_value="9"), mock.patch("builtins.print"):
+            result = tools.dispatch("ask_user", json.dumps({"question": "Pick one", "options": ["A", "B"]}))
+        self.assertEqual(result, "user answered: 9")
+
+    def test_empty_answer_keeps_turn_alive(self):
+        tools = self.tools()
+        with self.tty(), mock.patch("builtins.input", return_value="  "), mock.patch("builtins.print"):
+            result = tools.dispatch("ask_user", json.dumps({"question": "q?"}))
+        self.assertIn("no answer (empty)", result)
+        self.assertIn("proceed", result)
+
+    def test_eof_keeps_turn_alive(self):
+        tools = self.tools()
+        with self.tty(), mock.patch("builtins.input", side_effect=EOFError), mock.patch("builtins.print"):
+            result = tools.dispatch("ask_user", json.dumps({"question": "q?"}))
+        self.assertIn("no answer (input closed)", result)
+
+    def test_non_tty_refuses_without_blocking(self):
+        tools = self.tools()
+        fake_stdin = io.StringIO("")
+        fake_stdin.isatty = lambda: False
+        with mock.patch.object(sys, "stdin", fake_stdin), mock.patch("builtins.input") as asked:
+            result = tools.dispatch("ask_user", json.dumps({"question": "q?"}))
+        self.assertIn("no interactive user", result)
+        asked.assert_not_called()
+
+    def test_auto_mode_refuses(self):
+        tools = self.tools(auto=True)
+        with mock.patch("builtins.input") as asked:
+            result = tools.dispatch("ask_user", json.dumps({"question": "q?"}))
+        self.assertIn("autonomous mode", result)
+        asked.assert_not_called()
+
+    def test_background_worker_refuses(self):
+        tools = self.tools(persist=False)
+        with mock.patch("builtins.input") as asked:
+            result = tools.dispatch("ask_user", json.dumps({"question": "q?"}))
+        self.assertIn("background workers", result)
+        asked.assert_not_called()
+        deep = self.tools(persist=True, depth=1)
+        with mock.patch("builtins.input") as asked:
+            result = deep.dispatch("ask_user", json.dumps({"question": "q?"}))
+        self.assertIn("background workers", result)
+        asked.assert_not_called()
+
+    def test_options_capped(self):
+        tools = self.tools()
+        many = ["opt%d" % num for num in range(20)]
+        with self.tty(), mock.patch("builtins.input", return_value="1"), mock.patch("builtins.print") as shown:
+            result = tools.dispatch("ask_user", json.dumps({"question": "q?", "options": many}))
+        self.assertEqual(result, "user chose option 1: opt0")
+        printed = "\n".join(str(call.args[0]) for call in shown.call_args_list)
+        self.assertIn("12) opt11", printed)
+        self.assertNotIn("13) opt12", printed)
+
+
 class SysinfoTests(unittest.TestCase):
     def test_collect_all_checks_visible(self):
         report = tai.collect_sysinfo()
@@ -1514,7 +1630,7 @@ class LazyTests(unittest.TestCase):
 
     def test_core_minimal(self):
         self.assertEqual(self.names("hello"), set(tai.CORE_TOOLS))
-        self.assertEqual(len(tai.CORE_TOOLS), 8)
+        self.assertEqual(len(tai.CORE_TOOLS), 9)
 
     def test_synonym_triggers(self):
         pairs = [
@@ -3711,6 +3827,92 @@ class OpencodeParseTests(unittest.TestCase):
         self.assertEqual(json.loads(reply["tool_calls"][0]["arguments"]), {"command": "ls -lt"})
 
 
+class CliJsonRepairTests(unittest.TestCase):
+    def make_chat(self, tmp):
+        class FakeArgs:
+            profile = "t"
+            yes = True
+
+        os.environ["TAI_HOME"] = tmp
+        return tai.ChatClient(tai.Config(FakeArgs()))
+
+    def test_find_answer_none_on_prose(self):
+        self.assertIsNone(tai.opencode_find_answer("just some words"))
+        self.assertIsNone(tai.opencode_find_answer(""))
+        answer = {"content": "hi", "tool_calls": []}
+        self.assertEqual(tai.opencode_find_answer(json.dumps(answer)), answer)
+
+    def test_prose_then_json_repairs_once(self):
+        tools = [{"type": "function", "function": {"name": "shell"}}]
+        fixed = {"content": "checking", "tool_calls": [{"name": "shell", "arguments": '{"command": "ls"}'}]}
+        with tempfile.TemporaryDirectory() as tmp:
+            try:
+                chat = self.make_chat(tmp)
+                with mock.patch.object(tai.Opencode, "opencode_run", side_effect=["Ik kijk even mee op je scherm.", json.dumps(fixed)]) as runner:
+                    reply = chat.opencode_turn([{"role": "user", "content": "hi"}], tools, None, "opencode", "m")
+                self.assertEqual(runner.call_count, 2)
+                self.assertIn("no JSON object", runner.call_args_list[1][0][1])
+            finally:
+                os.environ.pop("TAI_HOME", None)
+        self.assertEqual(len(reply["tool_calls"]), 1)
+        self.assertEqual(reply["tool_calls"][0]["name"], "shell")
+
+    def test_prose_twice_accepts_second_as_final(self):
+        tools = [{"type": "function", "function": {"name": "shell"}}]
+        with tempfile.TemporaryDirectory() as tmp:
+            try:
+                chat = self.make_chat(tmp)
+                with mock.patch.object(tai.Opencode, "opencode_run", side_effect=["first prose", "second prose"]) as runner:
+                    reply = chat.opencode_turn([{"role": "user", "content": "hi"}], tools, None, "opencode", "m")
+                self.assertEqual(runner.call_count, 2)
+            finally:
+                os.environ.pop("TAI_HOME", None)
+        self.assertEqual(reply["content"], "second prose")
+        self.assertEqual(reply["tool_calls"], [])
+
+    def test_json_first_runs_no_repair(self):
+        tools = [{"type": "function", "function": {"name": "shell"}}]
+        answer = {"content": "checking", "tool_calls": [{"name": "shell", "arguments": '{"command": "ls"}'}]}
+        with tempfile.TemporaryDirectory() as tmp:
+            try:
+                chat = self.make_chat(tmp)
+                with mock.patch.object(tai.Opencode, "opencode_run", return_value=json.dumps(answer)) as runner:
+                    reply = chat.opencode_turn([{"role": "user", "content": "hi"}], tools, None, "opencode", "m")
+                self.assertEqual(runner.call_count, 1)
+            finally:
+                os.environ.pop("TAI_HOME", None)
+        self.assertEqual(len(reply["tool_calls"]), 1)
+
+    def test_no_repair_without_tools(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            try:
+                chat = self.make_chat(tmp)
+                with mock.patch.object(tai.Opencode, "opencode_run", return_value="just chatting") as runner:
+                    reply = chat.opencode_turn([{"role": "user", "content": "hi"}], None, None, "opencode", "m")
+                self.assertEqual(runner.call_count, 1)
+            finally:
+                os.environ.pop("TAI_HOME", None)
+        self.assertEqual(reply["content"], "just chatting")
+
+    def test_failed_repair_keeps_first_answer(self):
+        tools = [{"type": "function", "function": {"name": "shell"}}]
+        with tempfile.TemporaryDirectory() as tmp:
+            try:
+                chat = self.make_chat(tmp)
+                with mock.patch.object(tai.Opencode, "opencode_run", side_effect=["usable prose", tai.BackendError("boom")]) as runner:
+                    reply = chat.opencode_turn([{"role": "user", "content": "hi"}], tools, None, "opencode", "m")
+                self.assertEqual(runner.call_count, 2)
+            finally:
+                os.environ.pop("TAI_HOME", None)
+        self.assertEqual(reply["content"], "usable prose")
+
+    def test_prompt_contract_states_turn_ending(self):
+        self.assertIn("text without tool calls ends your turn", tai.DEFAULT_SYSTEM)
+        rules = tai.opencode_prompt([{"role": "user", "content": "hi"}], [])
+        self.assertIn("ends the turn at once", rules)
+        self.assertIn("Never narrate a planned action", rules)
+
+
 class RetrySummaryTests(unittest.TestCase):
     def test_empty_errors(self):
         self.assertEqual(tai.ChatClient.summarize_errors([]), "no details")