"""Project questions share their durable ask with every display lens.""" import ast import pathlib import re import json import os import pytest from ouroboros.gateway.history import _assemble_history_response from ouroboros.owner_quiz import quiz_states, record_asked, record_answered, reconcile_terminal from ouroboros.owner_wait import set_owner_wait from ouroboros.projects_registry import create_project from ouroboros.task_results import write_task_result, STATUS_RUNNING from tests.test_quiz_answer import _tool_ctx, _escalate, _decision_app, _post def test_real_escalation_bridge_history_detail_and_answer(tmp_path, monkeypatch): from supervisor import message_bus, state from ouroboros.gateway import tasks from starlette.applications import Starlette from starlette.routing import Route from starlette.testclient import TestClient project = create_project(tmp_path, "question-project", name="Question Project") write_task_result(tmp_path, "task-1", STATUS_RUNNING, project_id=project["id"], chat_id=project["chat_id"]) ctx = _tool_ctx(tmp_path, task_id="task-1", chat_id=project["chat_id"]) ctx.owner_wait_callback = lambda *_: None result = _escalate(ctx, question="Which storage?", wait_for_answer=True, stake="Where every later write lands.", options=[{"label": "Local", "detail": "Single-machine files"}, {"label": "Shared", "detail": "Multiple writers over network"}]) assert result.startswith("OK:") event = ctx.pending_events[0] qid = event["quiz_id"] assert quiz_states(tmp_path, "task-1")[qid]["option_details"] == [ "Single-machine files", "Multiple writers over network"] monkeypatch.setattr(message_bus, "DATA_DIR", tmp_path) state.init(tmp_path, 100) frames = [] bridge = message_bus.LocalChatBridge({}) bridge._broadcast_fn = frames.append monkeypatch.setattr(message_bus, "get_bridge", lambda: bridge) assert bridge.send_quiz(project["chat_id"], **{key: event[key] for key in ( "quiz_id", "question", "options", "stake", "assumption", "state", "task_id", "wait_for_answer")})[0] assert [row["type"] for row in frames] == ["quiz", "chat"] pointer = frames[1] assert pointer["system_type"] == "project_question_pointer" and pointer["chat_id"] == 1 # The live pointer frame is complete for display: the browser paints it without a detail read. assert pointer["question"] == "Which storage?" and pointer["options"] == ["Local", "Shared"] assert pointer["wait_for_answer"] is True and pointer["content"] == "Waiting for your answer in Question Project" # Main mirrors the Project's own form: the option details and the stake ride along too. details = ["Single-machine files", "Multiple writers over network"] assert pointer["option_details"] == details and pointer["stake"] == "Where every later write lands." assert not {"comment", "answered_index"} & pointer.keys() stored = (tmp_path / "logs/chat.jsonl").read_text().splitlines() assert len(stored) == 1 and json.loads(stored[0])["chat_id"] == project["chat_id"] main = json.loads(_assemble_history_response(tmp_path, 1, 10, 0))["messages"] assert len(main) == 1 and main[0]["quiz_id"] == qid assert main[0]["option_details"] == details and main[0]["stake"] == "Where every later write lands." assert not {"task_phase", "outcome_axes", "cost_final", "cancelable"} & main[0].keys() room = json.loads(_assemble_history_response(tmp_path, project["chat_id"], 10, 0))["messages"] assert room[0]["msg_type"] == "quiz" and room[0]["quiz"]["options"][1]["detail"] == "Multiple writers over network" monkeypatch.setattr(tasks, "request_drive_root", lambda request: tmp_path) app = Starlette(routes=[Route('/api/tasks/{task_id}', tasks.api_task_get)]) detail = TestClient(app).get('/api/tasks/task-1').json() assert detail["task_id"] == "task-1" and detail["project_id"] == project["id"] assert detail["owner_quiz"][qid]["option_details"] == quiz_states(tmp_path, "task-1")[qid]["option_details"] answered = _post(_decision_app(tmp_path, monkeypatch, live_task={"id": "task-1", "chat_id": project["chat_id"]}), {"request_id": "answer-1", "decision_id": f"quiz:task-1:{qid}", "option_index": 1}) assert answered.status_code == 200, answered.text main = json.loads(_assemble_history_response(tmp_path, 1, 10, 0))["messages"] assert main[0]["quiz_state"] == "answered" and main[0]["answered_index"] == 1 assert main[0]["question"] == "Which storage?" and main[0]["options"] == ["Local", "Shared"] assert main[0]["text"] == "You answered in Question Project" assert reconcile_terminal(tmp_path, "task-1") == [] assert quiz_states(tmp_path, "task-1")[qid]["option_details"][1] == "Multiple writers over network" def test_pointer_quota_and_dedup(tmp_path): from ouroboros.utils import append_jsonl project = create_project(tmp_path, "many-questions", name="Many Questions") for i in range(8): qid = f"q{i}" record_asked(tmp_path, "t1", quiz_id=qid, question="?", options=["a", "b"], wait_for_answer=True) row = {"type": "quiz", "direction": "out", "chat_id": project["chat_id"], "task_id": "t1", "ts": f"2026-09-09T00:00:0{i}Z", "text": "?", "quiz": {"quiz_id": qid, "options": ["a", "b"], "wait_for_answer": True}} append_jsonl(tmp_path / "logs/chat.jsonl", row) append_jsonl(tmp_path / "logs/chat.jsonl", row) append_jsonl(tmp_path / "logs/chat.jsonl", {**row, "quiz": {**row["quiz"], "quiz_id": "optional", "wait_for_answer": False}}) messages = json.loads(_assemble_history_response(tmp_path, 1, 3, 0))["messages"] assert len(messages) == 3 assert len({(row["task_id"], row["quiz_id"]) for row in messages}) == 3 assert all(row["system_type"] == "project_question_pointer" for row in messages) def test_an_optional_question_projects_into_main_with_its_assumption_and_never_reads_as_resumed(tmp_path, monkeypatch): """Every Project question is one Main row, whether or not the task waits on it. The optional one carries the assumption the task continues under and the asker's recommendation; a newer REQUIRED wait of the same task must not repaint it as a wait that ended, because it never waited (only a question the task waited on can have been resumed).""" from supervisor import message_bus, state project = create_project(tmp_path, "mixed-project", name="Mixed Project") write_task_result(tmp_path, "t1", STATUS_RUNNING, project_id=project["id"], chat_id=project["chat_id"]) ctx = _tool_ctx(tmp_path, task_id="t1", chat_id=project["chat_id"]) ctx.owner_wait_callback = lambda *_: None assert _escalate(ctx, question="Which format?", assumption="WebP meanwhile", options=[{"label": "PNG"}, {"label": "WebP", "recommended": True}]).startswith("OK:") optional = ctx.pending_events[0] monkeypatch.setattr(message_bus, "DATA_DIR", tmp_path) state.init(tmp_path, 100) frames = [] bridge = message_bus.LocalChatBridge({}) bridge._broadcast_fn = frames.append monkeypatch.setattr(message_bus, "get_bridge", lambda: bridge) assert bridge.send_quiz(project["chat_id"], **{key: optional[key] for key in ( "quiz_id", "question", "options", "stake", "assumption", "state", "task_id")})[0] assert [row["type"] for row in frames] == ["quiz", "chat"] live = frames[1] assert live["system_type"] == "project_question_pointer" and live["chat_id"] == 1 assert live["assumption"] == "WebP meanwhile" and live["recommended_index"] == 1 assert live["options"] == ["PNG", "WebP"] and "wait_for_answer" not in live assert live["content"] == "Unanswered · an answer is still accepted in Mixed Project" # The same task now waits on a newer required question. assert _escalate(ctx, question="Publish?", wait_for_answer=True, options=[{"label": "Yes", "recommended": True}, {"label": "No"}]).startswith("OK:") required = ctx.pending_events[1] assert bridge.send_quiz(project["chat_id"], **{key: required[key] for key in ( "quiz_id", "question", "options", "stake", "assumption", "state", "task_id", "wait_for_answer")})[0] set_owner_wait(tmp_path, "t1", {"quiz_id": required["quiz_id"], "wait_id": "w1", "state": "waiting"}) main = {row["quiz_id"]: row for row in json.loads(_assemble_history_response(tmp_path, 1, 10, 0))["messages"]} assert set(main) == {optional["quiz_id"], required["quiz_id"]} calm, waiting = main[optional["quiz_id"]], main[required["quiz_id"]] assert "owner_wait_state" not in calm and "wait_for_answer" not in calm assert calm["assumption"] == "WebP meanwhile" and calm["recommended_index"] == 1 assert calm["text"] == "Unanswered · an answer is still accepted in Mixed Project" assert waiting["owner_wait_state"] == "waiting" and waiting["recommended_index"] == 0 assert waiting["text"] == "Waiting for your answer in Mixed Project" # The Project room's own card of the optional question reads the same facts. room = {row["quiz"]["quiz_id"]: row["quiz"] for row in json.loads( _assemble_history_response(tmp_path, project["chat_id"], 10, 0))["messages"] if row.get("msg_type") == "quiz"} assert "owner_wait_state" not in room[optional["quiz_id"]] @pytest.mark.parametrize("same_timestamp", [False, True]) def test_activity_question_uses_same_memo_and_preserves_wait_semantics(tmp_path, monkeypatch, same_timestamp): from ouroboros.gateway import state as gs from ouroboros import utils from supervisor import queue project = create_project(tmp_path, "waiting-project", name="Waiting Project") write_task_result(tmp_path, "t1", STATUS_RUNNING, project_id=project["id"], root_phase_checkpoint={"post_task_synthesis": "running"}) record_asked(tmp_path, "t1", quiz_id="q1", question="?", options=["a", "b"], wait_for_answer=True, recommended_index=0, option_details=["Alpha", ""], stake="The rest of the run") set_owner_wait(tmp_path, "t1", {"quiz_id": "q1", "wait_id": "w1", "state": "waiting"}) monkeypatch.setattr(queue, "PENDING", []) monkeypatch.setattr(queue, "RUNNING", {"t1": {"task": {"id": "t1", "project_id": project["id"], "chat_id": project["chat_id"]}}}) gs._FINALIZING_MEMO.clear() reads = [] real = utils.read_json_dict monkeypatch.setattr(utils, "read_json_dict", lambda path: (reads.append(str(path)), real(path))[1]) rows = gs._chat_activities_snapshot_safe(tmp_path, direct_turns=[]) assert rows[0]["required_question"]["quiz_state"] == "open" assert rows[0]["required_question"]["text"] == "Waiting for your answer in Waiting Project" # The census pointer is as complete as the history row: the browser never paints a blank over it. assert rows[0]["required_question"]["question"] == "?" and rows[0]["required_question"]["options"] == ["a", "b"] assert rows[0]["required_question"]["option_details"] == ["Alpha", ""] assert rows[0]["required_question"]["stake"] == "The rest of the run" # A card painted from the census alone still badges the recommended option (index zero included). assert rows[0]["required_question"]["recommended_index"] == 0 # Folding an older Main card needs the named question's OWN asked_at, so the census # pointer carries it as `ts`. Losing that stamp does not fail loudly — it silently # stops every fold (web/modules/chat_decision.js::appendActivityQuestion), so the # cross-boundary contract is pinned on the producer side too. assert rows[0]["required_question"]["ts"] == quiz_states(tmp_path, "t1")["q1"]["asked_at"] assert reads.count(str(tmp_path / "task_results/t1.json")) == 1 gs._chat_activities_snapshot_safe(tmp_path, direct_turns=[]) assert reads.count(str(tmp_path / "task_results/t1.json")) == 1 path = tmp_path / "task_results/t1.json" before = path.stat() set_owner_wait(tmp_path, "t1", {"quiz_id": "q1", "wait_id": "w1", "state": "resumed"}, "w1") if same_timestamp: # Atomic replacement within one filesystem timestamp tick can retain # both mtime and size: waiting/resumed have the same serialized length. os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns)) after = path.stat() assert after.st_ino != before.st_ino assert (after.st_mtime_ns, after.st_size) == (before.st_mtime_ns, before.st_size) previous_reads = reads.count(str(path)) rows = gs._chat_activities_snapshot_safe(tmp_path, direct_turns=[]) assert rows[0]["required_question"]["owner_wait_state"] == "resumed" assert rows[0]["required_question"]["text"] == "Unanswered · the task continued; an answer is still accepted in Waiting Project" assert reads.count(str(path)) == previous_reads + 1 gs._chat_activities_snapshot_safe(tmp_path, direct_turns=[]) assert reads.count(str(path)) == previous_reads + 1 assert quiz_states(tmp_path, "t1")["q1"]["state"] == "open" # A wait that ended on its OWN bound resumed without an answer, so the # question is still wanted: no new wait state, just the additive reason. set_owner_wait(tmp_path, "t1", {"quiz_id": "q1", "wait_id": "w1", "state": "resumed", "resume_reason": "timeout"}, "w1") gs._FINALIZING_MEMO.clear() rows = gs._chat_activities_snapshot_safe(tmp_path, direct_turns=[]) pointer = rows[0]["required_question"] assert pointer["text"] == "Unanswered · the task continued; an answer is still accepted in Waiting Project" assert pointer["owner_wait_resume_reason"] == "timeout" def test_details_are_immutable_optional_and_length_checked(tmp_path): record_asked(tmp_path, "t", quiz_id="new", question="?", options=["a", "b"], option_details=["A", "B"]) record_asked(tmp_path, "t", quiz_id="new", question="?", options=["a", "b"], option_details=["changed", "changed"]) record_answered(tmp_path, "t", quiz_id="new", option_index=1, request_id="r") record_asked(tmp_path, "t", quiz_id="old", question="?", options=["a", "b"]) reconcile_terminal(tmp_path, "t") assert quiz_states(tmp_path, "t")["new"]["option_details"] == ["A", "B"] assert "option_details" not in quiz_states(tmp_path, "t")["old"] with pytest.raises(ValueError): record_asked(tmp_path, "t", quiz_id="bad", question="?", options=["a", "b"], option_details=["A"]) def test_question_presentation_shared_fixture(): """Both sides of the parity fixture: Python emits exactly the fixture row for each case, and the browser (web/tests/question_presentation.test.js) reads the fixture status out of that row.""" from pathlib import Path from ouroboros.project_dialogue import project_question_pointer cases = json.loads((Path(__file__).resolve().parents[1] / 'web/tests/fixtures/question_presentation_parity.json').read_text(encoding="utf-8")) keys = ("quiz_state", "owner_wait_state", "owner_wait_resume_reason", "wait_for_answer", "wait_ended_at", "answered_index", "comment", "source_status") for case in cases: block = {"quiz_id": "q", **case["block"]} if case["block"] is not None else None wait = {"quiz_id": "q", **case["owner_wait"]} if case["owner_wait"] is not None else None pointer = project_question_pointer( {"task_id": "task", "text": "Which?", "quiz": {"quiz_id": "q", "wait_for_answer": True, "options": ["a", "b"]}}, block, {"id": "p", "chat_id": 12, "name": "Project"}, wait) assert pointer["text"] == case["status"] + " in Project", case["case"] assert {key: pointer[key] for key in keys if key in pointer} == case["row"], case["case"] assert pointer["question"] == "Which?" and pointer["options"] == ["a", "b"] # A legacy label-only ask retained no details; Main says so exactly as the Project card does. assert "option_details" not in pointer and "stake" not in pointer narrow = project_question_pointer({"task_id": "task", "quiz_id": "q", "wait_for_answer": True}, {"quiz_id": "q", "state": "open"}, {"id": "p", "chat_id": 12, "name": "Project"}, None) assert not {"question", "options", "option_details", "stake"} & narrow.keys(), \ "unknown display fields are omitted, never blanked" def test_project_room_quiz_row_carries_the_wait_facts(tmp_path): """A wait the owner resumed by ordinary input leaves no frame behind: the replayed card reads the task's wait record like the Main pointer does.""" from ouroboros.utils import append_jsonl project = create_project(tmp_path, "resumed-project", name="Resumed Project") write_task_result(tmp_path, "t1", STATUS_RUNNING, project_id=project["id"], chat_id=project["chat_id"]) record_asked(tmp_path, "t1", quiz_id="q1", question="Which?", options=["a", "b"], wait_for_answer=True) append_jsonl(tmp_path / "logs/chat.jsonl", {"type": "quiz", "direction": "out", "chat_id": project["chat_id"], "task_id": "t1", "ts": "2026-09-16T00:00:00Z", "text": "Which?", "quiz": {"quiz_id": "q1", "options": ["a", "b"], "wait_for_answer": True}}) room = json.loads(_assemble_history_response(tmp_path, project["chat_id"], 10, 0))["messages"] assert "owner_wait_state" not in room[0]["quiz"] set_owner_wait(tmp_path, "t1", {"quiz_id": "q1", "wait_id": "w1", "state": "waiting"}) room = json.loads(_assemble_history_response(tmp_path, project["chat_id"], 10, 0))["messages"] assert room[0]["quiz"]["owner_wait_state"] == "waiting" set_owner_wait(tmp_path, "t1", {"quiz_id": "q1", "wait_id": "w1", "state": "resumed"}) room = json.loads(_assemble_history_response(tmp_path, project["chat_id"], 10, 0))["messages"] assert room[0]["quiz"]["owner_wait_state"] == "resumed" and room[0]["quiz"]["wait_for_answer"] is True main = json.loads(_assemble_history_response(tmp_path, 1, 10, 0))["messages"] assert main[0]["text"] == "Unanswered · the task continued; an answer is still accepted in Resumed Project" set_owner_wait(tmp_path, "t1", {"quiz_id": "q2", "wait_id": "w2", "state": "waiting"}) room = json.loads(_assemble_history_response(tmp_path, project["chat_id"], 10, 0))["messages"] assert room[0]["quiz"]["owner_wait_state"] == "resumed" from tests.test_contracts import REPO_ROOT, _dict_type_discriminator, _dict_literal_keys def _string_tuple(node: ast.AST) -> tuple[str, ...] | None: if isinstance(node, ast.Tuple) and node.elts and all( isinstance(elt, ast.Constant) and isinstance(elt.value, str) for elt in node.elts): return tuple(elt.value for elt in node.elts) return None def _function_node(source_path: pathlib.Path, name: str) -> ast.AST: tree = ast.parse(source_path.read_text(encoding="utf-8")) found = [node for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name] assert len(found) == 1, f"expected one {name} in {source_path.name}" return found[0] def test_project_question_pointer_display_fields_share_one_contract(): """Main mirrors a Project question as the Project's own form, so each pointer producer and both ABI mirrors carry the same display fields, ``option_details`` and ``stake`` included. The live frame (``message_bus.send_quiz``) copies them in a key loop and the census (``state._task_activity_facts``) in a key tuple, neither of which the envelope literal scan above can see; the browser keeps what it merges in ``chat_decision.js`` MIRROR_FIELDS.""" from typing import get_args, get_origin, get_type_hints from ouroboros.gateway.contracts import ChatOutbound from ouroboros.project_dialogue import project_question_pointer hints = get_type_hints(ChatOutbound, include_extras=True) assert "option_details" in hints and "stake" in hints # Both optional: a legacy label-only ask retained no details, and most asks name no stake. for field in ("option_details", "stake"): assert str(hints[field]).startswith(("typing.NotRequired[", "typing_extensions.NotRequired[")), hints[field] details = get_args(hints["option_details"])[0] assert get_origin(details) is list and get_args(details) == (str,) assert get_args(hints["stake"]) == (str,) api_types = (REPO_ROOT / "web" / "modules" / "api_types.js").read_text(encoding="utf-8") typedef = re.search(r"@typedef \{Object\} ChatOutbound\b(?P.*?)\n \*/", api_types, re.S) assert typedef, "api_types.js missing ChatOutbound" assert re.search(r"@property \{string\[\]=\} option_details\b", typedef.group("body")) assert re.search(r"@property \{string=\} stake\b", typedef.group("body")) mirror = re.search(r"const MIRROR_FIELDS = \[([^\]]*)\]", (REPO_ROOT / "web" / "modules" / "chat_decision.js").read_text(encoding="utf-8")) assert mirror, "chat_decision.js MIRROR_FIELDS moved" browser_fields = set(re.findall(r"'([a-z_]+)'", mirror.group(1))) assert {"option_details", "stake"} <= browser_fields assert browser_fields <= set(hints), sorted(browser_fields - set(hints)) # History: the producer itself emits every field the browser merges. block = {"quiz_id": "q", "state": "open", "question": "Which?", "options": ["a", "b"], "option_details": ["A detail", ""], "stake": "What rides on it", "assumption": "a", "recommended_index": 0, "host_facts": "Asked by task t, origin unknown."} pointer = project_question_pointer({"task_id": "t", "quiz_id": "q"}, block, {"id": "p", "chat_id": 12, "name": "Project"}, None) assert browser_fields <= set(pointer), sorted(browser_fields - set(pointer)) assert pointer["option_details"] == ["A detail", ""] and pointer["stake"] == "What rides on it" assert pointer["host_facts"] == "Asked by task t, origin unknown." # A block that recorded no host sentence yields a pointer without the key, never an empty one. bare = project_question_pointer({"task_id": "t", "quiz_id": "q"}, {**block, "host_facts": ""}, {"id": "p", "chat_id": 12, "name": "Project"}, None) assert "host_facts" not in bare # Live delivery: the frame literal plus its copied key loop, all declared in ChatOutbound. send_quiz = _function_node(REPO_ROOT / "supervisor" / "message_bus.py", "send_quiz") frames = [node for node in ast.walk(send_quiz) if isinstance(node, ast.Dict) and _dict_type_discriminator(node) == "chat"] assert len(frames) == 1, "send_quiz builds one pointer frame" literal, unknown = _dict_literal_keys(frames[0]) assert not unknown copied = [_string_tuple(node.iter) for node in ast.walk(send_quiz) if isinstance(node, ast.For) and _string_tuple(node.iter) and "recommended_index" in _string_tuple(node.iter)] assert len(copied) == 1, "send_quiz copies the pointer's display fields in one key loop" live = literal | set(copied[0]) assert {"option_details", "stake"} <= set(copied[0]) assert browser_fields <= live, sorted(browser_fields - live) assert live <= set(hints), sorted(live - set(hints)) # Census: the quiz facts the stat-keyed memo keeps for required_question. facts = _function_node(REPO_ROOT / "ouroboros" / "gateway" / "state.py", "_task_activity_facts") census = [_string_tuple(gen.iter) for node in ast.walk(facts) if isinstance(node, ast.DictComp) for gen in node.generators if _string_tuple(gen.iter) and "question" in _string_tuple(gen.iter)] assert len(census) == 1, "_task_activity_facts keeps the quiz display fields in one key tuple" quiz_sourced = browser_fields - {"project_name"} assert quiz_sourced <= set(census[0]), sorted(quiz_sourced - set(census[0])) def _census_row_with_quiz_wait(tmp_path, monkeypatch, *, project_id: str = "", chat_id: int = 0): from ouroboros.gateway import state as gs from supervisor import queue write_task_result(tmp_path, "t1", STATUS_RUNNING, project_id=project_id, root_phase_checkpoint={"post_task_synthesis": "running"}) record_asked(tmp_path, "t1", quiz_id="q1", question="?", options=["a", "b"], wait_for_answer=True) set_owner_wait(tmp_path, "t1", {"quiz_id": "q1", "wait_id": "w1", "state": "waiting"}) monkeypatch.setattr(queue, "PENDING", []) monkeypatch.setattr(queue, "RUNNING", {"t1": {"task": {"id": "t1", "project_id": project_id, "chat_id": chat_id}}}) gs._FINALIZING_MEMO.clear() return gs def test_unreadable_question_detail_is_disclosed_per_row_never_read_as_unblocked(tmp_path, monkeypatch): """A row that MAY be blocked on an owner answer must not project as free of questions.""" from ouroboros import project_dialogue project = create_project(tmp_path, "waiting-project", name="Waiting Project") gs = _census_row_with_quiz_wait(tmp_path, monkeypatch, project_id=project["id"], chat_id=project["chat_id"]) def _boom(*_a, **_k): raise RuntimeError("quiz detail unreadable") monkeypatch.setattr(project_dialogue, "project_question_pointer", _boom) rows = gs._chat_activities_snapshot_safe(tmp_path, direct_turns=[]) assert rows[0]["required_question_unavailable"] is True assert "required_question" not in rows[0] # The rest of the census stays intact: the row is still there with its own live phase. assert rows[0]["phase"] in {"working", "finalizing"} def test_quiz_wait_without_a_project_pointer_is_disclosed_not_dropped(tmp_path, monkeypatch): """A recorded owner-question wait whose Project pointer cannot be built is unknown, not absent.""" gs = _census_row_with_quiz_wait(tmp_path, monkeypatch, project_id="", chat_id=0) rows = gs._chat_activities_snapshot_safe(tmp_path, direct_turns=[]) assert rows[0]["required_question_unavailable"] is True assert "required_question" not in rows[0] def test_readable_question_detail_carries_no_unavailable_flag(tmp_path, monkeypatch): project = create_project(tmp_path, "waiting-project", name="Waiting Project") gs = _census_row_with_quiz_wait(tmp_path, monkeypatch, project_id=project["id"], chat_id=project["chat_id"]) rows = gs._chat_activities_snapshot_safe(tmp_path, direct_turns=[]) assert rows[0]["required_question"]["quiz_state"] == "open" assert "required_question_unavailable" not in rows[0]