diff --git a/CLAUDE.md b/CLAUDE.md index 2079044..5993fe7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -27,7 +27,7 @@ MCP data tools and web search. User-facing docs are in `README.md`; this file is drafts), `commands.py` (/credits), `access.py` (who is an admin). `features/`: `holding_news.py`, `movers.py`, `dm_buttons.py`, `post.py`, `periods/` (local `period_returns` tool, cache-first; `windows.py`, `rows.py`, `backfill.py` = PERIOD_BACKFILL loop). - `app.py` wiring/startup; `context.py` `Ctx` (settings, store, backend, registry, bot). - `lib/` holds only what the bot needs to run. Developer helpers live in `tools/`: `report.py` (`make prompt-report`), - `capture.py` (real MCP output for fixtures); the bot never imports them. + `capture.py` (real MCP output for fixtures), `period_cache_without_rowid.py` (one-off DB conversion); the bot never imports them. ## Request flow `handlers.on_message` logs the message -> `_trigger` (mention / reply-to-bot / DM / movers list) -> diff --git a/lib/llm/gate.py b/lib/llm/gate.py index 6f05d10..b80dbc0 100644 --- a/lib/llm/gate.py +++ b/lib/llm/gate.py @@ -34,9 +34,6 @@ LIVE_WORDS = re.compile( r"\b(news|today|latest|yesterday|tonight|this week|right now|breaking|announce\w*|who won|" r"score|weather|happening|headlines?|current\w*|updates?|recent\w*|just (now|in))\b", re.I) -# Asking about a trading session ("what happened at close"): answers must be fetched, not recalled. -SESSION_WORDS = re.compile( - r"\b(?:(?:at|the|into|after) (?:the )?(?:close|open)|pre-?market|after[- ]hours|closing|opening)\b", re.I) @dataclass(frozen=True) @@ -68,10 +65,6 @@ def is_portfolio_question(text: str, st: Settings) -> bool: return bool(PORTFOLIO_WEAK.search(text)) and not _has_ticker(text) -def is_session_question(text: str) -> bool: - return bool(SESSION_WORDS.search(text)) - - def route(text: str, st: Settings, registry: Registry, *, force_full: bool = False) -> Route: """Tools for this message. With TOKEN_SAVER off, or `force_full` (a search/data request such as a movers list, a preset button or a retry after NEEDS_TOOLS), everything is offered.""" @@ -80,7 +73,7 @@ def route(text: str, st: Settings, registry: Registry, *, force_full: bool = Fal return Route(up, st.search) portfolio = is_portfolio_question(text, st) market = portfolio or bool(MARKET_WORDS.search(text)) or _has_ticker(text) - live = is_session_question(text) or bool(LIVE_WORDS.search(text)) + live = bool(LIVE_WORDS.search(text)) servers = [s for s in up if (portfolio if s.cfg.get("gate") == "portfolio" else market)] search = st.search and (market or live) simple = not (servers or search) diff --git a/lib/llm/xai.py b/lib/llm/xai.py index 7e5c6b9..477514d 100644 --- a/lib/llm/xai.py +++ b/lib/llm/xai.py @@ -37,8 +37,11 @@ def answer_text(resp) -> str: """The response's last message. `output_text` joins every message, and with server-side search the model can write a draft, search, then write the answer, all in one response.""" messages = [i for i in resp.output if i.type == "message"] - texts = ["".join(getattr(c, "text", "") or "" for c in m.content if c.type == "output_text") - for m in messages] + parts = [[getattr(c, "text", "") or "" for c in m.content if c.type == "output_text"] for m in messages] + texts = ["".join(p) for p in parts] + for p in parts: + if len([t for t in p if t.strip()]) > 1: # not handled yet: this line tells us if it happens + log.info("xAI message had %d text parts (%s chars)", len(p), "+".join(str(len(t)) for t in p)) if len([t for t in texts if t.strip()]) > 1: log.info("xAI response had %d messages; kept the last (items: %s)", len(texts), " ".join(i.type for i in resp.output)) diff --git a/lib/prompts.py b/lib/prompts.py index dcfda7a..d9cb3da 100644 --- a/lib/prompts.py +++ b/lib/prompts.py @@ -69,7 +69,7 @@ def down_note(down: list[MCPServer]) -> str: def chat_prompt(*, transcript: str, sender: str, private: bool, reply_quote: str | None, msg_id: int, now: str, down: str = "", saver: bool = True, middle: str = "", - fresh: bool = False, session: bool = False) -> str: + fresh: bool = False) -> str: """The user turn: transcript, the replied-to message, and the volatile tail (kept last so the cached prefix before it is unchanged).""" head = (f"Private one-to-one chat with {sender}, not the group. Transcript:\n{transcript}" @@ -89,9 +89,6 @@ def chat_prompt(*, transcript: str, sender: str, private: bool, reply_quote: str "for anything else asked, such as prices or a comparison with market data); don't reuse " "numbers or refusals from earlier replies. When comparing portfolios, or asked \"why\", " "write 2-3 short paragraphs on the tailwinds and headwinds of each portfolio, not a summary.") - if session: # a trading-session question: news and reasons in earlier replies are stale by now - tail += (" Search for current news before answering, and say why each stock moved. Don't reuse " - "news or reasons from earlier messages in the transcript.") return head + tail + down diff --git a/lib/store.py b/lib/store.py index b58850d..08e01d8 100644 --- a/lib/store.py +++ b/lib/store.py @@ -38,7 +38,7 @@ -- for that window. movers = the holdings that moved most, as JSON [[code, gain, gain_pct], ...]. CREATE TABLE IF NOT EXISTS period_cache ( portfolio_id INTEGER, start TEXT, end TEXT, pct REAL, gain REAL, currency TEXT, movers TEXT, - v INTEGER, PRIMARY KEY (portfolio_id, start, end)); + v INTEGER, PRIMARY KEY (portfolio_id, start, end)) WITHOUT ROWID; """ PERIOD_CACHE_VERSION = 1 # bump when the stored shape changes: rows of another version are dropped @@ -77,6 +77,7 @@ def __init__(self, path: str): self._lock = threading.RLock() self.bot_id: int | None = None # set once the bot has started (DMs are stored per bot) self._run("PRAGMA journal_mode=WAL") + self._run("PRAGMA synchronous=NORMAL") # with WAL: no fsync per commit; a power cut may lose the last few with self._lock: self._conn.executescript(SCHEMA) # Drop "NOTHING" digests sent by an earlier version, so they aren't fed to the model. diff --git a/lib/telegram/handlers.py b/lib/telegram/handlers.py index 8ab1550..b4f669e 100644 --- a/lib/telegram/handlers.py +++ b/lib/telegram/handlers.py @@ -18,7 +18,7 @@ from ..features import dm_buttons, holding_news, movers, periods, post from ..history import compact, format_rows from ..llm.base import BUSY_REPLY, is_busy -from ..llm.gate import Route, is_portfolio_question, is_session_question, route +from ..llm.gate import Route, is_portfolio_question, route from ..msgtext import describe, sender_name from ..prompts import MOVERS_EXPLAIN, PRESET_PROMPT, chat_prompt, down_note from ..store import is_dm @@ -179,8 +179,7 @@ async def _gate_text(self, msg: Message, trig: Trigger) -> str: text += " " + compact(prev.text)[:FOLLOW_UP_CHARS] return text - async def _chat_text(self, msg: Message, trig: Trigger, middle: str = "", fresh: bool = False, - session: bool = False) -> str: + async def _chat_text(self, msg: Message, trig: Trigger, middle: str = "", fresh: bool = False) -> str: ctx, st = self.ctx, self.ctx.st rows = await asyncio.to_thread(ctx.store.history, msg.chat_id, st.history_limit) transcript = format_rows(rows, ctx.self_name, st.tz, line_max=st.history_line_max, @@ -188,8 +187,7 @@ async def _chat_text(self, msg: Message, trig: Trigger, middle: str = "", fresh: return chat_prompt( transcript=transcript, sender=sender_name(msg), private=trig.private, reply_quote=self._quote(trig), msg_id=msg.message_id, now=ctx.now(), - down=down_note(ctx.registry.down()), saver=st.token_saver, middle=middle, fresh=fresh, - session=session) + down=down_note(ctx.registry.down()), saver=st.token_saver, middle=middle, fresh=fresh) # -- answering --------------------------------------------------------------------- async def _answer(self, msg: Message, trig: Trigger) -> None: @@ -259,11 +257,9 @@ async def _generate(self, msg: Message, trig: Trigger, draft: Draft | None) -> s if chosen.simple: # the model must not be told it has no tools chosen = Route([], False) fresh = is_portfolio_question(context, st) and any(s.cfg.get("gate") == "portfolio" for s in chosen.servers) - session = is_session_question(context) and not chosen.simple - parts = [{"type": "text", "text": await self._chat_text(msg, trig, fresh=fresh, session=session)}] + parts = [{"type": "text", "text": await self._chat_text(msg, trig, fresh=fresh)}] parts += await self._images(msg, trig) answer = await ask(ctx, parts, chosen, - must_search=session and chosen.search, # a session recap must not be recalled cache_key=f"chat-{chat_id}", on_text=on_text, chat_id=chat_id, extra_tools=extra, admin=admin) return format_answer(answer.text, movers_bots=st.movers_bots, summary=trig.movers) diff --git a/tests/test_handlers.py b/tests/test_handlers.py index e31610c..ba7f5a5 100644 --- a/tests/test_handlers.py +++ b/tests/test_handlers.py @@ -90,14 +90,6 @@ async def test_partial_route_that_needs_more_is_asked_again_with_everything(env, assert "NEEDS_TOOLS" in backend.seen[0]["system"] # the partial route told the model how to ask -async def test_session_question_forces_a_search(env, store): - ctx, backend, h = make_ctx(env, store, [step("fresh recap", searches=1)], servers=[FakeMcp("yahoo")]) - m = user_msg(ctx.bot, "@stockbot what happened at close?") - await run(h, m) - assert backend.seen[0]["choice"] == "required" and backend.seen[0]["search"] is True - assert "Search for current news" in backend.seen[0]["prompt"] - - async def test_market_question_offers_tools_up_front(env, store): mcp = FakeMcp("yahoo") ctx, backend, h = make_ctx(env, store, [step("ok")], servers=[mcp]) diff --git a/tests/test_llm_backends.py b/tests/test_llm_backends.py index 79a25b2..f205dd5 100644 --- a/tests/test_llm_backends.py +++ b/tests/test_llm_backends.py @@ -247,6 +247,17 @@ async def test_xai_keeps_only_the_last_message_when_it_drafts_before_searching() assert seen == ["draft", "answer"] +async def test_xai_logs_a_message_with_several_text_parts_and_still_joins_them(caplog): + from types import SimpleNamespace as N + + from lib.llm.xai import answer_text + caplog.set_level("INFO", logger="bot") + message = N(type="message", content=[N(type="output_text", text="one"), N(type="output_text", text="two")]) + assert answer_text(N(output=[message], output_text="onetwo")) == "onetwo" + assert [r.getMessage() for r in caplog.records if "text parts" in r.getMessage()] == [ + "xAI message had 2 text parts (3+3 chars)"] + + async def test_xai_failed_response_at_capacity_is_overloaded_other_failures_are_not(): from types import SimpleNamespace as N st = config.load({"TELEGRAM_BOT_TOKEN": "1:x", "XAI_API_KEY": "k"}) diff --git a/tests/test_period_cache_tool.py b/tests/test_period_cache_tool.py new file mode 100644 index 0000000..4102099 --- /dev/null +++ b/tests/test_period_cache_tool.py @@ -0,0 +1,25 @@ +import sqlite3 + +from lib.store import Store +from tools.period_cache_without_rowid import convert + +OLD = """CREATE TABLE period_cache ( + portfolio_id INTEGER, start TEXT, end TEXT, pct REAL, gain REAL, currency TEXT, movers TEXT, + v INTEGER, PRIMARY KEY (portfolio_id, start, end))""" + + +def test_an_old_period_cache_keeps_its_rows_and_becomes_without_rowid(tmp_path): + path = str(tmp_path / "x.db") + db = sqlite3.connect(path) + db.execute(OLD) + db.executemany("INSERT INTO period_cache VALUES (?,?,?,?,?,?,?,?)", + [(1, f"2026-01-0{d}", f"2026-01-0{d}", 1.5, 10.0, "AUD", "[]", 1) for d in range(1, 4)]) + db.commit() + db.close() + assert convert(path) == 3 + assert convert(path) is None # already converted + db = sqlite3.connect(path) + assert "WITHOUT ROWID" in db.execute("SELECT sql FROM sqlite_master WHERE name = 'period_cache'").fetchone()[0] + assert db.execute("SELECT COUNT(*) FROM period_cache").fetchone()[0] == 3 + db.close() + assert Store(path).period_count(1, "2026-01-01", "2026-01-31") == 3 # the bot opens it fine diff --git a/tests/test_runner.py b/tests/test_runner.py index 5c6e16d..17b3d38 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -242,22 +242,6 @@ def test_gate_live_question_gets_search_only(settings): assert r.search and not r.servers -@pytest.mark.parametrize("text", ["what happened at close?", "how did it go at the open", "any pre-market moves"]) -def test_gate_market_session_words_are_live(text, settings): - y, s = servers() - r = route(text, settings, registry(y, s)) - assert r.search and not r.simple - - -def test_session_question_prompt_demands_current_news(): - from lib.llm.gate import is_session_question - from lib.prompts import chat_prompt - assert is_session_question("what happened at close?") and not is_session_question("hello") - kw = dict(transcript="t", sender="A", private=False, reply_quote=None, msg_id=1, now="now") - assert "Search for current news" in chat_prompt(**kw, session=True) - assert "Search for current news" not in chat_prompt(**kw) - - def test_gate_full_when_forced_or_token_saver_off(settings, env): y, s = servers() assert route("hello", settings, registry(y, s), force_full=True).servers == [y, s] diff --git a/tests/test_store.py b/tests/test_store.py index 397dde5..658b5f2 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -109,3 +109,9 @@ def test_period_flush_by_date_range_removes_every_window_that_overlaps_it(tmp_pa row("2023-01-01", "2023-12-31"), row("2025-01-07", "2025-01-13")]) assert s.period_flush(7, "2024-01-01", "2024-12-31") == 3 # the week spanning New Year both ways, and June assert sorted(s.period_get(7, "2000-01-01", "2100-01-01")) == [("2023-01-01", "2023-12-31"), ("2025-01-07", "2025-01-13")] + + +def test_store_uses_wal_with_normal_sync_and_a_without_rowid_period_cache(tmp_path): + st = Store(str(tmp_path / "x.db")) + assert st._run("PRAGMA synchronous") == [(1,)] + assert "WITHOUT ROWID" in st._run("SELECT sql FROM sqlite_master WHERE name = 'period_cache'")[0][0] diff --git a/tools/period_cache_without_rowid.py b/tools/period_cache_without_rowid.py new file mode 100644 index 0000000..201b928 --- /dev/null +++ b/tools/period_cache_without_rowid.py @@ -0,0 +1,45 @@ +"""One-off: convert an existing database's period_cache to WITHOUT ROWID (stop the bot first).""" + +from __future__ import annotations + +import sqlite3 +import sys + +# Keep in step with the period_cache table in lib/store.py. +CREATE = """CREATE TABLE period_cache ( + portfolio_id INTEGER, start TEXT, end TEXT, pct REAL, gain REAL, currency TEXT, movers TEXT, + v INTEGER, PRIMARY KEY (portfolio_id, start, end)) WITHOUT ROWID""" + + +def convert(path: str) -> int | None: + """Rebuild period_cache keeping every row. Returns the row count, or None if there was nothing to do.""" + db = sqlite3.connect(path, isolation_level=None) + try: + row = db.execute("SELECT sql FROM sqlite_master WHERE name = 'period_cache'").fetchone() + if row is None or "WITHOUT ROWID" in row[0]: + return None + db.execute("BEGIN") + db.execute("ALTER TABLE period_cache RENAME TO period_cache_old") + db.execute(CREATE) + db.execute("INSERT INTO period_cache SELECT * FROM period_cache_old") + count = db.execute("SELECT COUNT(*) FROM period_cache").fetchone()[0] + db.execute("DROP TABLE period_cache_old") + db.execute("COMMIT") + db.execute("VACUUM") + return count + finally: + db.close() + + +def main(argv: list[str]) -> int: + if len(argv) != 2: + print("usage: python -m tools.period_cache_without_rowid (stop the bot first)") + return 2 + count = convert(argv[1]) + print("period_cache is already WITHOUT ROWID (or absent); nothing to do" if count is None + else f"converted period_cache: {count} rows kept") + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv))