diff --git a/docs/architecture.md b/docs/architecture.md index dc03bc0..d032835 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -36,6 +36,8 @@ serialization while using explicit sessions; they do not swap agent globals. | Workspace | Root, bound tool adapters, graph, GitHub client and launch context | | AgentSession | Actor, conversation, durable task scope, interaction controller, learning feedback and subagents | | Turn | Response state, callbacks, capability token, selected model and request usage | +| ConversationHistory | Versioned private turn entries and provider/display projections, scoped by actor, interaction workspace and actual session ID | +| EventsRepository | Transactional journal append, idempotent entry IDs, expected-head checks and private record reads over the existing SQLite `events` table | Session identity includes actor, interaction scope, session ID and workspace root. Personal memory retains its existing global workspace scope. Source snapshots and @@ -43,6 +45,24 @@ project history use the source scope. Changing projects selects another adapter existing sessions retain their original workspace. Context variables carry session, event, approval and execution state through provider threads without global mutation. +ConversationHistory records user input, each received provider response, correlated +typed tool results, interaction references and turn outcomes. Entry payloads are +versioned defensive JSON snapshots; private rows are not returned by generic event, +SSE or interface APIs. The runtime records before the next dispatch and stops if a +journal write fails. Workspace snapshot nodes carry a private journal checkpoint; +rollback appends a branch-selection entry and preserves the abandoned branch. +Legacy message lists are imported with provenance without parsing their text. Saved +provider continuations remain bound to their original provider/model and cannot replay +tools when that binding no longer validates. + +The provider projection restores typed responses and correlated tool replies, and +fails if a native call/reply group or continuation binding is incomplete. The display +projection renders each completed turn once. Context compaction remains a derived +message view and never rewrites the journal. Context usage reports estimated, +content-free component attribution, a stable-prefix fingerprint, newly introduced +history, retransmission and duplicate-entry counts; provider-reported input totals +remain separate from those estimates. + ## Package responsibilities | Package | Responsibility | diff --git a/docs/history-and-rollback.md b/docs/history-and-rollback.md index c6d4823..75fce45 100644 --- a/docs/history-and-rollback.md +++ b/docs/history-and-rollback.md @@ -1,6 +1,6 @@ # Conversation history and rollback -History preserves conversation snapshots as a tree. Each node is tied to a user actor, workspace, session, and parent version. The current head represents the active line. Creating a new change after a rollback can leave the old branch available for inspection rather than discarding it. +Workspace history preserves snapshots as a tree. Each node is tied to a user actor, workspace, session, and parent version, and its private checkpoint points to the canonical conversation journal. The journal retains full versioned provider responses and tool results independently of display snapshots. Rollback appends a branch selection; abandoned conversation entries remain stored and recoverable. ## Inspect history @@ -10,7 +10,7 @@ Use `/history` in the terminal UI or the session-history routes listed in [tool- In the TUI, choose a numbered history node to preview the target. Explicitly confirm with `/rollback NUMBER confirm` after reviewing it. The session head is updated only through the rollback operation. A stale head or a mismatched session is rejected instead of overwriting a newer branch. In the web API, the rollback endpoint is scoped to the requested session and node; authenticate and inspect its preview/result before continuing work. -Rollback restores the recorded files in scope and rewinds the selected conversation history. It does not reverse external effects such as messages, network requests, deployments, provider charges, or database changes outside the captured workspace. It also does not imply that shared memory claims, agent credentials, or another session's state have been erased. Check those stores separately. +Rollback restores the recorded files in scope and selects the conversation branch captured by that node. It does not reverse external effects such as messages, network requests, deployments, provider charges, or database changes outside the captured workspace. It also does not imply that shared memory claims, agent credentials, or another session's state have been erased. Check those stores separately. ## Recovery points and safety diff --git a/docs/index.md b/docs/index.md index 7745051..39d682a 100644 --- a/docs/index.md +++ b/docs/index.md @@ -43,6 +43,7 @@ This index covers the current product guides, developer references, and dated en - [V3 plan index](superpowers/plans/index.md) links the issue-specific implementation plans and identifies their coverage. - [V3 execution receipt implementation plan](superpowers/plans/2026-10-14-v3-issue-236.md) records receipt identity, explicit task binding, structured effect verification, compatibility, and validation for issue #236. - [V3 runtime events implementation plan](superpowers/plans/2026-10-14-v3-issue-255.md) records the canonical event contract, correlation, interface adapters, and validation for issue #255. +- [V3 canonical conversation history plan](superpowers/plans/2026-10-15-v3-issue-247.md) records the private journal, runtime integration, history checkpoints, context accounting, and validation for issue #247. - Machine-readable audit receipts and benchmark inputs remain beside their corresponding reports under `docs/` and `docs/benchmarks/`. The repository's [English overview](../README.md) links here. Technical guides are maintained in English; README translations provide localized entry points. diff --git a/docs/providers.md b/docs/providers.md index 5d6c022..ec5334e 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -110,6 +110,15 @@ fixture does not establish live gateway acceptance. Offline fixtures and installed SDK types validate the contract; they do not establish live provider interoperability. +Conversation history is owned by the runtime. It records full structured responses, +correlated tool results and provider-bound continuation state in private journal +entries; public runtime events remain redacted projections. Native requests restore +typed call/reply groups only when every reply and the saved provider/model binding +validate. After restart, a mismatched continuation is reported and pauses native +projection rather than silently rebinding or executing a tool again. See +[architecture](architecture.md) for journal scope, rollback branching and context +accounting behavior. + ## Optional dependencies `pip install -e '.[web]'` installs FastAPI and Uvicorn. Provider extras are `.[claude]`, `.[gemini]`, `.[perplexity]`, `.[bedrock]`, and `.[vertex]`; `.[cloud]` groups the cloud integrations. `.[browser]` installs Playwright; browser execution also needs a browser installation. `.[all]` installs the supported optional Python integrations. The official installer has its own pinned set of dependencies; see [installation](installation.md). diff --git a/docs/superpowers/plans/2026-10-15-v3-issue-247.md b/docs/superpowers/plans/2026-10-15-v3-issue-247.md new file mode 100644 index 0000000..2181ae4 --- /dev/null +++ b/docs/superpowers/plans/2026-10-15-v3-issue-247.md @@ -0,0 +1,85 @@ +# V3 Issue #247: Canonical conversation history + +**Issue:** [#247 — Canonical conversation history](https://github.com/EvanProgramming/OpenKyrozen/issues/247) +**Requirements:** R101–R105, R107, R113, R114, R191 +**Branch:** `Evan/v3-247-conversation-history`, based on `main` at `d71d2bd42b68485feb4571d3aad490d7d8327aa3` +**Delivery:** one GPG-signed PR; leave the PR unmerged and issue open. + +## Scope and ownership + +The runtime owns turn recording and projections. Reuse the SQLite `events` table through private repository methods, with a `ConversationHistory` service for immutable scoped records, projections, forks, and import reconciliation. Keep workspace snapshots in `HistoryManager`; link snapshot nodes to journal cursors transactionally rather than adding another history store. Provider continuation remains private and bound to the adapter/model that created it. Interfaces consume display projections, and generic event/SSE APIs never expose journal payloads. AgentEngine replacement, structured compaction redesign, and unrelated V3 issues are out of scope. + +The implementation path currently spans `openkyrozen/persistence/events.py` and `store.py`, `openkyrozen/agent/runtime.py` and `loop.py`, `action_rounds.py`, `initial_response.py`, `response_recovery.py`, `prompts.py`, context accounting, `openkyrozen/workspace/history.py`, and the CLI/TUI/Web chat adapters. Existing typed contracts live in `openkyrozen/providers/models.py`; public runtime events live in `openkyrozen/agent/events.py` and remain redacted projections. + +## Indexed implementation plan + +### Task 1 — Private journal repository and immutable entry model + +- [x] Add regressions for append idempotency, conflicting duplicate IDs, stale expected heads, deep defensive snapshots, unsupported versions, scope isolation, and generic event/API privacy. +- [x] Extend the existing event repository with transaction-scoped conversation append/read/head methods. Use unique entry IDs and expected-head checks; do not expose journal rows through generic event listing. +- [x] Add versioned immutable entries and `ConversationHistory` with user/workspace/session scope, append, provider/display projections, fork/branch, and explicit legacy import operations. +- [x] Add lossless codecs for `ModelResponse`, `ToolReply`, and `ToolResult`; validate continuation provider/model binding before native projection. + +### Task 2 — Runtime-owned recording and restart behavior + +- [x] Record user input, actual model responses, correlated tool results, interaction/plan references, and turn outcomes exactly once; fail closed before any subsequent dispatch if persistence fails. +- [x] Reconcile legacy `AgentRuntime.chat()` message lists through an explicit import adapter without deleting or rewriting the prior branch. +- [x] Restore scoped history on session initialization, including the actual CLI/default session ID. Never infer historic protocol actions by parsing prose and never execute restored calls. +- [x] Add restart/branch tests, including a saved continuation with a changed adapter/model that pauses native projection explicitly without rebinding or replay. + +### Task 3 — Workspace history linkage and conversation projection + +- [x] Link workspace snapshot nodes to the journal cursor in the same transaction as history-node creation; rollback appends branch selection while preserving abandoned journal records and existing filesystem recovery behavior. +- [x] Replace accumulated action-round summaries with one evolving projection. Record each provider response and result once; repairs append only their correction. Keep routing, learning, compaction, and delegation requests auxiliary. +- [x] Retain legacy list shapes and compaction as derived views; native projections preserve complete call/reply groups or fail safely. + +### Task 4 — Context assembly, integrations, and evidence + +- [x] Route CLI/TUI/Web user/final recording through runtime projections; remove duplicate interface writes while preserving display and wire contracts. +- [x] Build deterministic stable/dynamic/conversation prompt sections, remove automatic graph-context injection, and preserve registry schemas and the existing one-inventory-per-request legacy path. +- [x] Extend `context_usage()` with content-free attribution, prefix fingerprint, new/retransmitted history and duplicate-entry counts while preserving existing fields and separating estimates from provider totals. +- [x] Update architecture/provider/history docs and tests. Run Python 3.12/3.13 focused and full validation, browser tests with installed Chromium, Go TUI and offline acceptance; record all skips. +- [ ] Obtain independent review, fix material findings, verify signed commit locally and on GitHub, attach PR, and wait for final-head CI/reviews. + +## Validation record + +### Post-review fixes on PR #283 + +- [x] Reproduce and fix seven new Codex findings from review of `1112102`: typed history request projection, occurrence-based duplicate metrics, rollback conflict translation, journal-only session discovery, textless failure projection, empty canonical Web branch authority, and in-loop tool inventory refresh. +- [x] Reproduce and fix three additional Codex findings: typed-message context preflight, failure-path cache refresh, and preventing a truncated display cache from replacing canonical history. +- [x] Fix the matching CodeRabbit findings and isolate the Web journal test from shared database/session state. Preserve the explicit restart policy: mismatched provider/model continuation fails closed instead of being converted to text or rebound. +- [x] Re-run focused regressions and both supported Python full suites; repository checks passed. Final-head CI and review-thread resolution remain pending publication. +- [x] Reproduce and fix four further final-head findings: safely compact legacy dictionary prefixes ahead of native history; pin final provider continuations even when they have no tool calls; include model-free turn outcomes in provider projection without duplicating model responses; and keep the canonical journal authoritative over older Web snapshots. +- Follow-up signature check: GitHub reports the prior head commit `d20fbf7` as `bad_email`; the account's published GPG key marks `piggyevan123@gmail.com` verified and the noreply address unverified. A corrected follow-up uses the verified email; the existing commit history is not rewritten. + +PASS — Python 3.12 and 3.13 focused conversation-history, rollback, interaction, Web, streaming, and native-provider suites (92 tests per focused run; one unrelated browser skip in the initial focused runs). + +PASS — Full Python 3.12 and 3.13 suites: `KYROZEN_BROWSER_TESTS=1 make test PYTHON=.venv/bin/python` and `.venv313/bin/python` (833 tests each, including browser session/auth tests). + +PASS — Python 3.13 `make check`, `make lint`, `make docs-check`, and `make tui-test`; `git diff --check`. + +PASS — Latest focused compaction, conversation-history, native-provider, provider-contract, and Web suites: 106 tests on Python 3.13. + +PASS — Latest full Python 3.12 and 3.13 suites: 837 tests each with `KYROZEN_BROWSER_TESTS=1`; browser session/auth coverage ran. + +PASS — Latest `make check`, `make lint`, `make docs-check`, `make tui-test`, and `git diff --check`. `make check` and `make docs-check` require the project virtual environment on `PATH`; the system Python lacked the installed `requests` dependency. + +PASS — Four final-head review findings reproduced and fixed. Their review threads still need replies and resolution after the signed commit is published. + +PASS — Seven Codex findings reproduced and fixed: rollback session refresh, nested branch projection, legacy snapshot backfill, authoritative Web journal projection, early acceptance finalization, atomic rollback notifications, and per-dispatch context metrics (including typed native messages). Added regressions pass. + +PASS — Post-review regressions: canonical provider projection and call/reply correlation, repeated journal identity accounting, stale rollback error translation, journal-only session discovery without payload exposure, textless error projection, empty canonical Web branch, and tool inventory replacement/fingerprint invalidation. + +PASS — Final Python 3.12 and 3.13 full suites: 833 tests each, including browser auth/session tests with `KYROZEN_BROWSER_TESTS=1`. An earlier Python 3.12 run hit a transient 5-second headless-startup timeout; the isolated test and subsequent complete run passed. + +PASS — Context-preflight, failed-turn cache, truncation, saved-model binding, and Web test-isolation regressions pass. Mismatched provider/model continuations are explicitly rejected, preserving the no-rebind restart policy. + +PASS — Python 3.13 `make check`, `make lint`, `make docs-check`, and `make tui-test`; `git diff --check`. + +PASS — CodeRabbit reviewed head `2f0c746` and posted four findings; three were fixed. Its automatic recovery suggestion was declined because converting/rebinding mismatched continuation state would violate the selected fail-closed restart policy. Codex posted three further findings; all are fixed. Threads remain open until the fixes are published. + +PASS — Two additional Codex findings on `469ca5a` were reproduced and fixed: explicit legacy replacement now commits its branch marker and replacement entries in one head-checked transaction, and context history metrics remap only retained message identities after preflight or overflow compaction. Atomic-failure and compacted-history regressions pass. + +## Compatibility and risks + +Legacy callers keep their message-list shape through an explicit import adapter. Private journal rows contain full conversation and continuation data and must remain unavailable to generic event APIs, logs, and interface projections. If a saved native continuation cannot be validated against its live adapter/model, the request pauses for explicit recovery; tools are never automatically replayed. diff --git a/docs/superpowers/plans/2026-10-16-v3-issue-242.md b/docs/superpowers/plans/2026-10-16-v3-issue-242.md index 55894f8..f8950a2 100644 --- a/docs/superpowers/plans/2026-10-16-v3-issue-242.md +++ b/docs/superpowers/plans/2026-10-16-v3-issue-242.md @@ -45,4 +45,3 @@ Check storage concurrency under `BEGIN IMMEDIATE`, in-memory rollback on all exc ## Validation record Implementation is complete on `Evan/v3-242-atomic-plan-edits`. Python 3.12 and 3.13 each passed all 815 tests with browser tests enabled. `make check`, `make lint`, `make docs-check`, `make tui-test`, `make tui-acceptance`, `make agent-acceptance`, and `make subagent-acceptance` passed. Focused plan and migration suites passed (46 tests). Independent review found and verified fixes for typed operation schemas, atomic scheduling claims, and rejection of mixed legacy/canonical arguments. No paid provider calls or new dependencies were used. The repository and issue were refreshed before creating this plan; `main` was `d71d2bd42b68485feb4571d3aad490d7d8327aa3`, issue #242 remains open, and PR #283 remains open and unmerged. - diff --git a/docs/superpowers/plans/index.md b/docs/superpowers/plans/index.md index 79275d3..f13b27c 100644 --- a/docs/superpowers/plans/index.md +++ b/docs/superpowers/plans/index.md @@ -14,6 +14,7 @@ - [V3 #245 — Versioned, user-approved Strategic Plans](2026-10-16-v3-issue-245.md) - [V3 #236 — Execution receipts and task association](2026-10-14-v3-issue-236.md) - [V3 #255 — Shared typed runtime events](2026-10-14-v3-issue-255.md) +- [V3 #247 — Canonical conversation history](2026-10-15-v3-issue-247.md) Add a linked issue-specific plan here before coding any later V3 issue. diff --git a/openkyrozen/agent/__init__.py b/openkyrozen/agent/__init__.py index 83146ba..f177164 100644 --- a/openkyrozen/agent/__init__.py +++ b/openkyrozen/agent/__init__.py @@ -1,5 +1,6 @@ """OpenKyrozen agent module.""" from .events import RuntimeEvent +from .conversation import ConversationEntry, ConversationHistory -__all__ = ["RuntimeEvent"] +__all__ = ["RuntimeEvent", "ConversationEntry", "ConversationHistory"] diff --git a/openkyrozen/agent/action_rounds.py b/openkyrozen/agent/action_rounds.py index 724225d..df40b75 100644 --- a/openkyrozen/agent/action_rounds.py +++ b/openkyrozen/agent/action_rounds.py @@ -37,6 +37,9 @@ def _execute_action_rounds(self, turn): consecutive_search_failures = 0 # track failed search_web calls to prevent loops total_search_calls = sum(record["action"] == "search_web" for record in turn.tool_records) unsupported_action_retries = 0 + evolving_messages = None + previous_model_reply = None + previous_tool_results: list[str] = [] distribution_reply = None coordinator = self.subagent_manager.coordinator delegated_this_turn = coordinator and any(run["version"] != turn.agent_versions.get(run["run_id"]) @@ -198,6 +201,14 @@ def _execute_action_rounds(self, turn): memory = self._build_memory_context(turn.user_input, context=turn.memory_context) if memory: summary_messages.insert(2, {"role": "system", "content": memory}) + if evolving_messages is not None: + summary_messages = evolving_messages + summary_messages.append({"role": "assistant", "content": previous_model_reply or ""}) + if previous_tool_results: + summary_messages.append({"role": "user", "content": "Tool results:\n" + "\n".join(previous_tool_results)}) + if error_hint or search_throttle: + summary_messages.append({"role": "user", "content": error_hint + search_throttle + + self._build_task_progress_hint()}) if distribution_reply is not None: step_reply, distribution_reply = distribution_reply, None else: @@ -216,6 +227,8 @@ def _execute_action_rounds(self, turn): # Observe this response once. The same parsed result drives task # updates, unknown-action handling, loop control, and final rendering. step_reply, step_meta = self._observe_turn_response(turn, step_reply, summary_messages) + if step_meta["define_tool_registered"]: + self._refresh_tool_inventory(summary_messages) step_reply, step_meta, plan_recovery_exhausted = self._recover_plan_response(turn, step_reply, step_meta, summary_messages, inspection_complete=bool(turn.tool_records), @@ -243,6 +256,8 @@ def _execute_action_rounds(self, turn): turn.protocol_error_message = self._UNSUPPORTED_ACTION_PROTOCOL_MESSAGE break step_reply = self._call_llm_with_spinner(summary_messages + [{ + "role": "assistant", "content": step_reply, + }, { "role": "user", "content": ( f"System: {self._UNSUPPORTED_ACTION_PROTOCOL_MESSAGE} " @@ -254,6 +269,8 @@ def _execute_action_rounds(self, turn): if not step_reply: continue step_reply, step_meta = self._observe_turn_response(turn, step_reply, summary_messages) + if step_meta["define_tool_registered"]: + self._refresh_tool_inventory(summary_messages) next_tool_calls = step_meta["tool_calls"] if step_meta["protocol_error"]: turn.protocol_error_message = step_meta["protocol_error"] @@ -289,6 +306,8 @@ def _execute_action_rounds(self, turn): break # Re-observe the replacement response exactly once. step_reply, step_meta = self._observe_turn_response(turn, step_reply, summary_messages) + if step_meta["define_tool_registered"]: + self._refresh_tool_inventory(summary_messages) next_tool_calls = step_meta["tool_calls"] if step_meta["protocol_error"]: turn.protocol_error_message = step_meta["protocol_error"] @@ -351,13 +370,18 @@ def _execute_action_rounds(self, turn): if next_pending_desc: nudge += f"The next task is: \"{next_pending_desc}\". " nudge += "Output the Action block for this task NOW. Do not explain or summarise — just act." - summary_messages.append({"role": "user", "content": nudge}) + summary_messages.extend([ + {"role": "assistant", "content": step_reply}, + {"role": "user", "content": nudge}, + ]) step_reply = self._call_llm_with_spinner(summary_messages).strip() turn.turn_prompt_total += self._last_prompt_tokens turn.turn_completion_total += self._last_completion_tokens if not step_reply: break step_reply, step_meta = self._observe_turn_response(turn, step_reply, summary_messages) + if step_meta["define_tool_registered"]: + self._refresh_tool_inventory(summary_messages) next_tool_calls = step_meta["tool_calls"] if step_meta["protocol_error"]: turn.protocol_error_message = step_meta["protocol_error"] @@ -375,17 +399,19 @@ def _execute_action_rounds(self, turn): turn.final_answer = step_meta["clean"] or self._deterministic_tool_summary(turn.tool_records) break else: - summary_messages.append({ + summary_messages.extend([{"role": "assistant", "content": step_reply}, { "role": "user", "content": "System: You have not output an Action block. " "Output the next JSON Action block now to continue." - }) + }]) step_reply = self._call_llm_with_spinner(summary_messages).strip() turn.turn_prompt_total += self._last_prompt_tokens turn.turn_completion_total += self._last_completion_tokens if not step_reply: break step_reply, step_meta = self._observe_turn_response(turn, step_reply, summary_messages) + if step_meta["define_tool_registered"]: + self._refresh_tool_inventory(summary_messages) next_tool_calls = step_meta["tool_calls"] if step_meta["protocol_error"]: turn.protocol_error_message = step_meta["protocol_error"] @@ -431,6 +457,9 @@ def _execute_action_rounds(self, turn): "requires args in format path|content", ]) for r in next_results ) + evolving_messages = summary_messages + previous_model_reply = step_reply + previous_tool_results = list(next_results) # Hard limit: too many search_web calls → one last chance to complete if total_search_calls >= 7 and consecutive_search_failures >= 3: final_msg = ( @@ -441,7 +470,8 @@ def _execute_action_rounds(self, turn): "directly. Acknowledge any gaps honestly." ).format(consecutive_search_failures) step_reply = self._call_llm_with_spinner( - summary_messages + [{"role": "user", "content": final_msg}] + summary_messages + [{"role": "assistant", "content": step_reply}, + {"role": "user", "content": final_msg}] ).strip() step_reply, search_meta = self._observe_turn_response(turn, step_reply, summary_messages) interaction_reply = self._interaction_gate(search_meta, turn.user_input) diff --git a/openkyrozen/agent/compaction.py b/openkyrozen/agent/compaction.py index 73e29eb..60e2253 100644 --- a/openkyrozen/agent/compaction.py +++ b/openkyrozen/agent/compaction.py @@ -82,8 +82,22 @@ def _category(message: dict[str, Any], index: int, last_non_system: int) -> str: return "conversation" -def estimate_messages(messages: list[dict[str, Any]]) -> tuple[int, dict[str, int]]: +def _accounting_message(message: Any) -> dict[str, Any]: + """Create a text-only estimate without changing native provider history.""" + from openkyrozen.providers.models import ModelResponse, ToolReply + + if isinstance(message, ModelResponse): + calls = [{"name": call.name, "arguments": call.arguments} for call in message.tool_calls] + content = message.text + ("\n" + str(calls) if calls else "") + return {"role": "assistant", "content": content} + if isinstance(message, ToolReply): + return {"role": "tool", "content": str(message.output)} + return message + + +def estimate_messages(messages: list[Any]) -> tuple[int, dict[str, int]]: """Estimate complete provider-visible prompt tokens and content categories.""" + messages = [_accounting_message(item) for item in messages] last_non_system = max((index for index, item in enumerate(messages) if item.get("role") != "system"), default=-1) breakdown: dict[str, int] = { @@ -101,7 +115,8 @@ def estimate_messages(messages: list[dict[str, Any]]) -> tuple[int, dict[str, in return total + 2, breakdown # small chat-framing allowance -def message_fingerprint(messages: list[dict[str, Any]]) -> str: +def message_fingerprint(messages: list[Any]) -> str: + messages = [_accounting_message(item) for item in messages] payload = "\n".join( f"{item.get('role', '')}\x00{item.get('content', '')}" for item in messages ) @@ -219,7 +234,7 @@ def _tool_parts(message: dict[str, Any]) -> tuple[str, list[str]] | None: return prefix + marker, trailing.splitlines() -def compact_for_pressure(messages: list[dict[str, Any]], state: ContextState, summarize: Summarizer, +def compact_for_pressure(messages: list[Any], state: ContextState, summarize: Summarizer, *, force: bool = False) -> CompactionResult: """Compact only under a declared-window pressure or an explicit overflow retry.""" working = list(messages) @@ -234,6 +249,19 @@ def under_pressure(current: dict[str, Any]) -> bool: status["compaction"] = {"status": "not_needed"} return CompactionResult(working) + from openkyrozen.providers.models import ModelResponse, ToolReply + if any(isinstance(item, (ModelResponse, ToolReply)) for item in working): + first_native = next(index for index, item in enumerate(working) + if isinstance(item, (ModelResponse, ToolReply))) + prefix = compact_for_pressure(working[:first_native], state, summarize, force=force) + working = prefix.messages + working[first_native:] + status = state.update(working, compaction=state.status.get("compaction")) + if prefix.impossible or under_pressure(status): + status["compaction"] = {"status": "native_history_cannot_be_compacted_safely"} + return CompactionResult(working, compacted=prefix.compacted, + history_digest=prefix.history_digest, impossible=True) + return CompactionResult(working, compacted=prefix.compacted, history_digest=prefix.history_digest) + non_system = [item for item in working if item.get("role") != "system" and not is_context_digest(item)] tracked_history = [item for item in non_system if id(item) in state.history_message_ids] history_protected = {id(item) for item in tracked_history[-KEEP_RECENT_MESSAGES:]} diff --git a/openkyrozen/agent/context.py b/openkyrozen/agent/context.py index a0fc737..11d9c81 100644 --- a/openkyrozen/agent/context.py +++ b/openkyrozen/agent/context.py @@ -1,6 +1,7 @@ from __future__ import annotations from typing import Any +import json from openkyrozen.agent.compaction import ContextState, compact_for_pressure, is_context_digest, message_fingerprint, retain_context_digests from openkyrozen.agent.types import ContextTooLargeError @@ -19,10 +20,95 @@ def context_usage(self) -> dict[str, Any] | None: def _store_context_status(self, state: ContextState) -> None: if not state.status: return + state.status.update({ + "component_estimates": dict(state.status.get("breakdown", {})), + "component_attribution_source": "estimated", + "stable_prefix_fingerprint": self._stable_prefix_info.get().get("fingerprint"), + "stable_prefix_tokens": self._stable_prefix_info.get().get("estimated_tokens", 0), + **self._context_history_metrics.get(), + }) with self._context_status_lock: self._context_status_by_scope[self._context_scope()] = dict(state.status) +def _record_context_history_metrics(self, messages: list[dict[str, Any]]) -> None: + from openkyrozen.agent.conversation import model_response_to_dict + from openkyrozen.providers.models import ModelResponse, ToolReply + + entries = self.conversation_history.active_entries() + visible: dict[tuple[str, str], int] = {} + message_keys: dict[int, tuple[str, str]] = {} + for index, message in enumerate(messages): + if isinstance(message, dict) and message.get("role") in {"user", "assistant"} and isinstance(message.get("content"), str): + key = (message["role"], message["content"]) + visible[key] = visible.get(key, 0) + 1 + elif isinstance(message, ModelResponse): + key = ("model.response", json.dumps(model_response_to_dict(message), sort_keys=True, ensure_ascii=False)) + visible[key] = visible.get(key, 0) + 1 + elif isinstance(message, ToolReply): + key = ("tool.result", json.dumps({"call_id": message.call_id, "name": message.name, + "output": message.output, "is_error": message.is_error}, + sort_keys=True, ensure_ascii=False)) + visible[key] = visible.get(key, 0) + 1 + else: + continue + message_keys[index] = key + entries_by_key: dict[tuple[str, str], list[str]] = {} + active_turn_id = self._conversation_turn_id.get() + for entry in entries: + if active_turn_id is not None and entry.turn_id == active_turn_id: + continue + payload = entry.payload + if entry.kind == "user.input": + key = ("user", payload.get("text", "")) + elif entry.kind in {"turn.outcome", "legacy.assistant"}: + key = ("assistant", payload.get("text", "")) + elif entry.kind == "model.response": + key = ("model.response", json.dumps(payload["response"], sort_keys=True, ensure_ascii=False)) + elif entry.kind == "tool.result": + key = ("tool.result", json.dumps({"call_id": payload.get("call_id"), "name": payload.get("name"), + "output": payload.get("output"), "is_error": payload.get("is_error", False)}, + sort_keys=True, ensure_ascii=False)) + else: + continue + entries_by_key.setdefault(key, []).append(entry.id) + projection = self._context_history_projection_ids.get() + if projection is not None and projection[0] == id(messages): + history_ids = [entry_id for _, entry_id in sorted(projection[1].items())] + start, end = projection[2] + mapped = set(projection[1]) + for index, key in message_keys.items(): + identities = entries_by_key.get(key, []) + if start <= index < end and index not in mapped and identities: + history_ids.append(identities[-1]) + else: + history_ids = [] + for key, count in visible.items(): + identities = entries_by_key.get(key, []) + if not identities: + continue + history_ids.extend(identities[:count]) + if count > len(identities): + history_ids.extend([identities[-1]] * (count - len(identities))) + seen = self._context_history_seen.get() + if seen is None: + seen = set() + self._context_history_seen.set(seen) + unique_history_ids = set(history_ids) + new_ids = unique_history_ids - seen + retransmitted_ids = unique_history_ids & seen + entry_tokens = {entry.id: max(1, (len(json.dumps(entry.payload, ensure_ascii=False)) + 3) // 4) + for entry in entries if entry.id in history_ids} + self._context_history_metrics.set({ + "new_entry_count": len(new_ids), + "retransmitted_entry_count": len(retransmitted_ids), + "duplicate_entry_count": len(history_ids) - len(set(history_ids)), + "new_context_tokens": sum(entry_tokens[item] for item in new_ids), + "retransmitted_history_tokens": sum(entry_tokens[item] for item in retransmitted_ids), + }) + seen.update(unique_history_ids) + + def _save_learning_context_digest(self, digest: dict[str, Any] | None) -> None: if (not digest or not is_context_digest(digest) or not self.memory_bank.session_id or not self._SELF_LEARNING_FLAGS.get("context_compression", True)): @@ -66,6 +152,27 @@ def _cache_reported_context_tokens(self, messages: list[dict], prompt_tokens: in counts.pop(next(iter(counts))) +def _update_context_history_projection(self, messages: list[Any], previous: list[Any]) -> None: + projection = self._context_history_projection_ids.get() + if projection is None or projection[0] != id(messages): + return + retained_positions = {} + for index, message in enumerate(messages): + retained_positions.setdefault(id(message), []).append(index) + remapped = {} + for old_index, entry_id in projection[1].items(): + if old_index >= len(previous): + continue + positions = retained_positions.get(id(previous[old_index]), []) + if positions: + remapped[positions.pop(0)] = entry_id + if remapped: + bounds = (min(remapped), max(remapped) + 1) + else: + bounds = (0, 0) + self._context_history_projection_ids.set((id(messages), remapped, bounds)) + + def _is_context_overflow_error(self, exc: Exception) -> bool: from openkyrozen.providers.errors import legacy_context_overflow return legacy_context_overflow(exc) @@ -99,11 +206,14 @@ def _prepare_context_for_call(self, messages: list[dict], model: str | None = No messages, state, lambda text, output_chars: self._summarize_context_with_chat_model(text, output_chars, model or state.model), ) + previous_messages = list(messages) messages[:] = result.messages + _update_context_history_projection(self, messages, previous_messages) if result.history_digest is not None: retained_ids = {id(item) for item in messages} retained_history = [item for item in self.short_term_memory if id(item) in retained_ids] self.short_term_memory = retain_context_digests([result.history_digest] + retained_history, self.SHORT_TERM_CAP * 2) + self.current_session.messages_compacted = True state.history_message_ids = {id(item) for item in self.short_term_memory} self._save_learning_context_digest(result.history_digest) self._store_context_status(state) @@ -125,11 +235,14 @@ def _recover_context_after_overflow(self, messages: list[dict], model: str | Non lambda text, output_chars: self._summarize_context_with_chat_model(text, output_chars, model or state.model), force=True, ) + previous_messages = list(messages) messages[:] = result.messages + _update_context_history_projection(self, messages, previous_messages) if result.history_digest is not None: retained_ids = {id(item) for item in messages} retained_history = [item for item in self.short_term_memory if id(item) in retained_ids] self.short_term_memory = retain_context_digests([result.history_digest] + retained_history, self.SHORT_TERM_CAP * 2) + self.current_session.messages_compacted = True state.history_message_ids = {id(item) for item in self.short_term_memory} self._save_learning_context_digest(result.history_digest) self._store_context_status(state) diff --git a/openkyrozen/agent/conversation.py b/openkyrozen/agent/conversation.py new file mode 100644 index 0000000..e9ddf39 --- /dev/null +++ b/openkyrozen/agent/conversation.py @@ -0,0 +1,310 @@ +"""Private, scoped conversation journal backed by the existing events table.""" +from __future__ import annotations + +import copy +import json +import math +import uuid +from dataclasses import dataclass, field +from typing import Any + +_MISSING = object() + + +class _ProviderProjection(list): + def __init__(self): + super().__init__() + self.entry_ids: list[str] = [] + + def append_entry(self, entry: ConversationEntry, message: Any) -> None: + self.append(message) + self.entry_ids.append(entry.id) + + +class ConversationHistoryError(ValueError): + """A journal write, read, or provider projection cannot be applied safely.""" + + +def _validate_json(value: Any) -> None: + if isinstance(value, dict): + if any(not isinstance(key, str) for key in value): + raise ConversationHistoryError("Conversation object keys must be strings") + for item in value.values(): + _validate_json(item) + elif isinstance(value, list): + for item in value: + _validate_json(item) + elif value is None or isinstance(value, (str, bool, int)): + return + elif not isinstance(value, float) or not math.isfinite(value): + raise ConversationHistoryError("Conversation payload must contain finite JSON values") + + +def _encode(value: Any) -> str: + _validate_json(value) + try: + return json.dumps(value, ensure_ascii=False, allow_nan=False, sort_keys=True, separators=(",", ":")) + except (TypeError, ValueError, RecursionError) as exc: + raise ConversationHistoryError("Conversation payload must be JSON serializable") from exc + + +@dataclass(frozen=True, init=False) +class ConversationEntry: + id: str + kind: str + parent_id: str | None + version: int + user_id: str + workspace_id: str + session_id: str + turn_id: str | None + request_id: str | None + call_id: str | None + task_id: str | None + created_at: str + _payload: str = field(repr=False) + + def __init__(self, *, id: str, kind: str, payload: Any, user_id: str, workspace_id: str, + session_id: str, parent_id: str | None = None, version: int = 1, + turn_id: str | None = None, request_id: str | None = None, + call_id: str | None = None, task_id: str | None = None, + created_at: str | None = None): + if not all(isinstance(value, str) and value.strip() for value in (id, kind, user_id, workspace_id, session_id)): + raise ConversationHistoryError("Conversation entry identity and scope fields must be nonempty") + if type(version) is not int or version != 1: + raise ConversationHistoryError("Unsupported conversation entry version") + from openkyrozen.persistence.models import utc_now + values = {"id": id, "kind": kind, "parent_id": parent_id, "version": version, + "user_id": user_id, "workspace_id": workspace_id, "session_id": session_id, + "turn_id": turn_id, "request_id": request_id, "call_id": call_id, + "task_id": task_id, "created_at": created_at or utc_now(), + "_payload": _encode(payload)} + for key, value in values.items(): + object.__setattr__(self, key, value) + + @property + def payload(self) -> Any: + return json.loads(self._payload) + + def as_dict(self) -> dict[str, Any]: + return {"id": self.id, "kind": self.kind, "parent_id": self.parent_id, + "version": self.version, "user_id": self.user_id, + "workspace_id": self.workspace_id, "session_id": self.session_id, + "turn_id": self.turn_id, "request_id": self.request_id, + "call_id": self.call_id, "task_id": self.task_id, + "created_at": self.created_at, "payload": self.payload} + + +class ConversationHistory: + def __init__(self, store, *, user_id: str = "local", workspace_id: str = "default", session_id: str): + self.store = store + self.user_id, self.workspace_id, self.session_id = user_id, workspace_id, session_id + + @property + def head(self) -> str | None: + return self.store.conversation_head(user_id=self.user_id, workspace_id=self.workspace_id, + session_id=self.session_id) + + def entry(self, kind: str, payload: Any, *, entry_id: str | None = None, **correlation) -> ConversationEntry: + parent_id = correlation.pop("parent_id", self.head) + return ConversationEntry(id=entry_id or f"conv_{uuid.uuid4().hex}", kind=kind, + payload=copy.deepcopy(payload), user_id=self.user_id, + workspace_id=self.workspace_id, session_id=self.session_id, + parent_id=parent_id, **correlation) + + def append(self, entry: ConversationEntry, *, expected_head: str | None | object = _MISSING) -> str: + if (entry.user_id, entry.workspace_id, entry.session_id) != (self.user_id, self.workspace_id, self.session_id): + raise ConversationHistoryError("Conversation entry belongs to a different scope") + if expected_head is _MISSING: + return self.store.append_conversation_entry(entry.as_dict()) + return self.store.append_conversation_entry(entry.as_dict(), expected_head=expected_head) + + def entries(self) -> tuple[ConversationEntry, ...]: + values = self.store.list_conversation_entries( + user_id=self.user_id, workspace_id=self.workspace_id, session_id=self.session_id, + ) + return tuple(ConversationEntry(**value) for value in values) + + def active_entries(self) -> tuple[ConversationEntry, ...]: + entries = self.entries() + targets = {item.payload.get("target_journal_head") for item in entries + if item.kind == "branch.selection" and item.payload.get("target_journal_head")} + active_at = {} + active = None + for item in entries: + if item.kind == "branch.selection": + target = item.payload.get("target_journal_head") + if target is not None and target not in active_at: + raise ConversationHistoryError("Branch selection references an unknown journal entry") + active = active_at.get(target) + else: + if item.payload.get("provenance") == "legacy_history_node" and item.payload.get("legacy_index") == 0: + active = None + active = (item, active) + if item.id in targets: + active_at[item.id] = active + result = [] + while active: + item, active = active + result.append(item) + result.reverse() + return tuple(result) + + def display_messages(self, *, exclude_turn_id: str | None = None) -> list[dict[str, Any]]: + messages = [] + for entry in self.active_entries(): + if exclude_turn_id is not None and entry.turn_id == exclude_turn_id: + continue + payload = entry.payload + if entry.kind == "user.input": + messages.append({"role": "user", "content": payload["text"]}) + elif entry.kind == "legacy.assistant": + messages.append({"role": "assistant", "content": payload["text"]}) + elif entry.kind == "turn.outcome": + text = payload.get("text") + if isinstance(text, str) and text: + messages.append({"role": "assistant", "content": text}) + return messages + + def provider_messages(self, *, provider_id: str, model: str, native: bool = True, + exclude_turn_id: str | None = None) -> list[Any]: + from openkyrozen.providers.models import ToolReply, validate_tool_history + messages = _ProviderProjection() + pending_calls = [] + entries = self.active_entries() + model_turn_text = {} + for entry in entries: + if entry.kind == "model.response" and entry.turn_id is not None: + model_turn_text[entry.turn_id] = model_response_from_dict(entry.payload["response"]).text + for entry in entries: + if exclude_turn_id is not None and entry.turn_id == exclude_turn_id: + continue + payload = entry.payload + if entry.kind == "user.input": + messages.append_entry(entry, {"role": "user", "content": payload["text"]}) + elif entry.kind == "legacy.assistant": + messages.append_entry(entry, {"role": "assistant", "content": payload["text"]}) + elif entry.kind == "model.response": + response = model_response_from_dict(payload["response"]) + if response.continuation is None and not response.tool_calls: + messages.append_entry(entry, {"role": "assistant", "content": response.text}) + else: + messages.append_entry(entry, response) + pending_calls.extend(response.tool_calls) + elif entry.kind == "tool.result": + call_id = payload.get("call_id") + if native and pending_calls: + messages.append_entry(entry, ToolReply( + call_id, payload["name"], payload["output"], payload.get("is_error", False), + )) + if pending_calls[0].id == call_id: + pending_calls.pop(0) + else: + output = payload.get("output", {}) + messages.append_entry(entry, {"role": "user", "content": ( + f"Tool result ({payload.get('name', 'tool')}): {output.get('text', '')}" + )}) + elif entry.kind == "turn.outcome": + text = payload.get("text") + if isinstance(text, str) and text and text != model_turn_text.get(entry.turn_id): + messages.append_entry(entry, {"role": "assistant", "content": text}) + validate_tool_history(messages, provider_id=provider_id, model=model, native=native) + return messages + + def import_legacy(self, messages: list[dict[str, Any]]) -> None: + """Import old display-only snapshots once; never infer controls from their text.""" + if self.head is not None: + return + entries = [] + parent_id = None + for index, message in enumerate(messages): + if not isinstance(message, dict) or message.get("role") not in {"user", "assistant"}: + continue + content = message.get("content", "") + if not isinstance(content, str): + continue + kind = "user.input" if message["role"] == "user" else "legacy.assistant" + entry = self.entry(kind, {"text": content, "provenance": "legacy_snapshot", "index": index}, + parent_id=parent_id) + entries.append(entry) + parent_id = entry.id + self.store.append_conversation_entries( + [entry.as_dict() for entry in entries], expected_head=None, + ) + + def reconcile_legacy(self, messages: list[dict[str, Any]]) -> None: + """Import an explicitly supplied replacement list as a new branch.""" + normalized = [item for item in messages if isinstance(item, dict) + and item.get("role") in {"user", "assistant"} + and isinstance(item.get("content", ""), str)] + if normalized == self.display_messages(): + return + parent = self.head + branch = self.entry("branch.selection", { + "target_journal_head": None, "reason": "legacy_message_list_replaced", + }, parent_id=parent) + entries = [branch] + parent_id = branch.id + for index, message in enumerate(normalized): + kind = "user.input" if message["role"] == "user" else "legacy.assistant" + entry = self.entry(kind, { + "text": message.get("content", ""), "provenance": "legacy_replacement", "index": index, + }, parent_id=parent_id) + entries.append(entry) + parent_id = entry.id + self.store.append_conversation_entries( + [entry.as_dict() for entry in entries], expected_head=parent, + ) + + +def model_response_to_dict(response) -> dict[str, Any]: + from openkyrozen.providers.models import ModelResponse + if not isinstance(response, ModelResponse): + raise TypeError("Expected ModelResponse") + continuation = response.continuation + return { + "text": response.text, + "tool_calls": [{"id": call.id, "name": call.name, "arguments": call.arguments} + for call in response.tool_calls], + "usage": copy.deepcopy(response.usage), + "finish_reason": response.finish_reason.value, + "metadata": copy.deepcopy(response.metadata), + "continuation": ({"provider_id": continuation.provider_id, + "request_model": continuation.request_model, + "protocol": continuation.protocol, + "payload": continuation.payload} if continuation else None), + } + + +def model_response_from_dict(value): + from openkyrozen.providers.models import ( + FinishReason, ModelResponse, ProviderContinuation, ToolCall, + ) + continuation = value.get("continuation") + return ModelResponse( + text=value["text"], + tool_calls=tuple(ToolCall(call["id"], call["name"], call["arguments"]) + for call in value.get("tool_calls", ())), + usage=value.get("usage"), finish_reason=FinishReason(value["finish_reason"]), + metadata=value.get("metadata", {}), + continuation=(ProviderContinuation(**continuation) if continuation else None), + ) + + +def tool_result_to_dict(result) -> dict[str, Any]: + from openkyrozen.tools.models import ToolResult + if not isinstance(result, ToolResult): + raise TypeError("Expected ToolResult") + error = result.error + return {"data": result.data, "text": result.text, + "error": ({"code": error.code, "message": error.message, + "retriable": error.retriable, "metadata": error.metadata} if error else None), + "metadata": result.metadata} + + +def tool_result_from_dict(value): + from openkyrozen.tools.models import ToolError, ToolResult + error = value.get("error") + return ToolResult(data=value.get("data"), text=value.get("text", ""), + error=ToolError(**error) if error else None, + metadata=value.get("metadata", {})) diff --git a/openkyrozen/agent/delegation_runtime.py b/openkyrozen/agent/delegation_runtime.py index 1754d1b..3d857bc 100644 --- a/openkyrozen/agent/delegation_runtime.py +++ b/openkyrozen/agent/delegation_runtime.py @@ -283,14 +283,33 @@ def _invoke_delegated(self, run, *, review, feedback, coordinator): if parent_publisher is not None: child_publisher = parent_publisher.fork(session_id=child.memory.session_id) event_token = self._runtime_event_publisher.set(child_publisher) + conversation_token = self._conversation_turn_id.set(f"turn_{child_id}_{uuid.uuid4().hex}") + record_token = self._record_foreground_model.set(True) + conversation_head_token = None started = time.monotonic() try: with self.use_session(child): - result = self._run_subagent_llm_result(profile, task, [], tools) + conversation = self.conversation_history + conversation_head_token = self._conversation_expected_head.set(conversation.head) + self._record_conversation_entry("user.input", {"text": task}) + try: + result = self._run_subagent_llm_result(profile, task, [], tools) + except Exception as exc: + self._record_conversation_entry("turn.outcome", { + "status": "failed", "error_type": type(exc).__name__, + }) + raise result["session_id"] = child_id result["metrics"] = self._subagent_usage_metrics(child_id, result.pop("started")) + self._record_conversation_entry("turn.outcome", { + "status": "completed", "text": result.get("result", ""), + }) return result finally: + if conversation_head_token is not None: + self._conversation_expected_head.reset(conversation_head_token) + self._record_foreground_model.reset(record_token) + self._conversation_turn_id.reset(conversation_token) with coordinator.lock: run.setdefault("usage", {})[child_id] = self._subagent_usage_metrics(child_id, started) self._active_usage_run_id.reset(usage_token) diff --git a/openkyrozen/agent/executor.py b/openkyrozen/agent/executor.py index 63ae2a1..bd0f9f1 100644 --- a/openkyrozen/agent/executor.py +++ b/openkyrozen/agent/executor.py @@ -260,7 +260,7 @@ def clean(item, remaining, depth=0): return value, remaining - len(value) return clean(value, budget)[0] - return ExecutionReceipt( + receipt = ExecutionReceipt( receipt_id=f"receipt_{uuid.uuid4().hex}", operation_id=self._operation_id(operation_scope, canonical, canonical_arguments if canonical_arguments is not None else args), @@ -294,6 +294,21 @@ def clean(item, remaining, depth=0): side_effects=getattr(expected_spec, "side_effects", "unknown"), verification=verification, ) + if getattr(self, "_conversation_turn_id", None) is not None and self._conversation_turn_id.get(): + from openkyrozen.agent.conversation import tool_result_to_dict + outcome = tool_result or ToolResult( + data=receipt.result_data, text=receipt.result, + error=(ToolError(receipt.failure or "execution_failed", receipt.result or "Tool failed") + if not receipt.success else None), + ) + self._record_conversation_entry("tool.result", { + "call_id": receipt.call_id, "name": receipt.action, + "arguments": receipt.arguments if receipt.arguments is not None else {"legacy": receipt.args}, + "output": tool_result_to_dict(outcome), "is_error": not outcome.success, + "receipt_id": receipt.receipt_id, + "verification": receipt.verification, + }, call_id=receipt.call_id, task_id=receipt.task_id) + return receipt def _run_tool(self, action: str, args: str, *, return_success: bool = False, diff --git a/openkyrozen/agent/loop.py b/openkyrozen/agent/loop.py index 0f080d4..f64ca47 100644 --- a/openkyrozen/agent/loop.py +++ b/openkyrozen/agent/loop.py @@ -2,7 +2,7 @@ import json from typing import Any -from openkyrozen.tasks.engine import is_terminal +from openkyrozen.tasks.engine import is_complete, is_terminal from openkyrozen.agent.modes import InteractionError, is_plan_acceptance from openkyrozen.workspace.history_port import TurnToken @@ -28,6 +28,12 @@ def _chat_turn(self, user_input: str, clear_tasks: bool = False, profile: str | previous_capability_token = self._execution_capability_token original_user_input = user_input history_token: TurnToken | None = None + conversation = self.conversation_history + if conversation.head is None: + conversation.import_legacy(list(self.short_term_memory)) + elif not getattr(self.current_session, "messages_compacted", False): + conversation.reconcile_legacy(list(self.short_term_memory)) + self._conversation_expected_head.set(conversation.head) self._interaction_controller.reconcile_plan_completion(self.tasks) try: history_token = self.history_manager().begin_turn( @@ -40,12 +46,14 @@ def _chat_turn(self, user_input: str, clear_tasks: bool = False, profile: str | user_id=self.memory_bank.user_id, workspace_id=self.memory_bank.workspace_id, session_id=self.memory_bank.session_id, ) + self._record_conversation_entry("user.input", {"text": original_user_input}) accepted_plan: dict[str, Any] | None = None mode_override = self._interaction_mode_override.get() state = self._interaction_controller.state(user_input) executing_plan = state.get("executing_plan") if mode_override is None else None pending_question = state.get("pending_question") if mode_override is None else None pending_plan = state.get("pending_plan") if mode_override is None else None + early_reply = None if mode_override is None and pending_question: resolved = self._interaction_controller.resolve_question( pending_question["request_id"], user_input, action="answer", @@ -56,21 +64,23 @@ def _chat_turn(self, user_input: str, clear_tasks: bool = False, profile: str | ) clear_tasks = False elif mode_override is None and is_plan_acceptance(user_input): + created = None try: accepted_plan, created = self._interaction_controller.accept_plan(self.tasks) except InteractionError: - return "There is no pending plan to accept." - if not created and not self._interaction_controller.state().get("executing_plan"): - return "Plan was already accepted." - user_input = ( - "Execute the accepted plan below in Agent mode. Complete each durable task with evidence.\n\n" - + json.dumps(accepted_plan, ensure_ascii=False) - ) - self._emit_stream_event({ - "event": "interaction", - "interaction": self._interaction_controller.envelope(user_input), - }) - clear_tasks = False + early_reply = "There is no pending plan to accept." + if created is False and not self._interaction_controller.state().get("executing_plan"): + early_reply = "Plan was already accepted." + if early_reply is None: + user_input = ( + "Execute the accepted plan below in Agent mode. Complete each durable task with evidence.\n\n" + + json.dumps(accepted_plan, ensure_ascii=False) + ) + self._emit_stream_event({ + "event": "interaction", + "interaction": self._interaction_controller.envelope(user_input), + }) + clear_tasks = False elif mode_override is None and pending_plan: user_input = ( f"Original request:\n{pending_plan.get('original_input', '')}\n\n" @@ -94,14 +104,28 @@ def _chat_turn(self, user_input: str, clear_tasks: bool = False, profile: str | } runtime.turn_start(user_input=original_user_input, **context) try: - reply = self._chat_turn_impl( - user_input, clear_tasks=clear_tasks, profile=profile, - memory_context=memory_context, - inspection_complete=bool(pending_question or pending_plan), - ) + if early_reply is not None: + self._turn_failed.set(False) + reply = early_reply + else: + reply = self._chat_turn_impl( + user_input, clear_tasks=clear_tasks, profile=profile, + memory_context=memory_context, + inspection_complete=bool(pending_question or pending_plan), + ) if backend := self._fast_used_backend.get(): reply = f"Made a decision with {'Jev' if backend == 'jev' else 'Kev'}.\n\n{reply}" except Exception as exc: + self._record_conversation_entry("turn.outcome", { + "status": "failed", "error_type": type(exc).__name__, + }) + from openkyrozen.agent.compaction import is_context_digest, retain_context_digests + display = conversation.display_messages() + limit = self.SHORT_TERM_CAP * 2 + digest = next((item for item in reversed(self.short_term_memory) if is_context_digest(item)), None) + projected = ([digest] if digest else []) + display + self.short_term_memory = retain_context_digests(projected, limit) + self.current_session.messages_compacted = bool(digest) or len(projected) > limit if active_plan: self._interaction_controller.complete_plan(active_plan, status="failed") self._track_fix_outcome(original_user_input, f"Turn failed: {type(exc).__name__}: {exc}") @@ -136,14 +160,33 @@ def _chat_turn(self, user_input: str, clear_tasks: bool = False, profile: str | self._emit_stream_event({"event": "interaction", "interaction": self.interaction_envelope()}) self._track_fix_outcome(original_user_input, reply) runtime.turn_end(reply=reply, success=True, **context) + interaction = self._interaction_controller.state() + self._record_conversation_entry("interaction.reference", { + "interaction": interaction, + "plan_id": interaction.get("pending_plan", {}).get("plan_id") + if isinstance(interaction.get("pending_plan"), dict) else None, + "task_ids": [item.get("id") for item in self.tasks.tasks if item.get("id")], + }) + self._record_conversation_entry("turn.outcome", { + "status": "failed" if self._turn_failed.get() else "completed", + "text": self._clean_final_response(reply), + }) + from openkyrozen.agent.compaction import is_context_digest, retain_context_digests + digest = next((item for item in reversed(self.short_term_memory) if is_context_digest(item)), None) + display = conversation.display_messages() + projected = ([digest] if digest else []) + display + truncated = len(projected) > self.SHORT_TERM_CAP * 2 + self.short_term_memory = retain_context_digests( + projected, self.SHORT_TERM_CAP * 2, + ) + self.current_session.messages_compacted = bool(digest) or truncated if history_token is not None: try: self.history_manager().commit_turn( history_token, user_message=original_user_input, assistant_message=reply, - conversation=list(self.short_term_memory) + [ - {"role": "user", "content": original_user_input}, - {"role": "assistant", "content": self._clean_final_response(reply)}, - ], interaction=self._interaction_controller.state(), tasks=list(self.tasks.tasks), + conversation=conversation.display_messages(), + interaction=self._interaction_controller.state(), tasks=list(self.tasks.tasks), + journal_head=conversation.head, ) except Exception as exc: self.memory_bank.store.append_event( diff --git a/openkyrozen/agent/models.py b/openkyrozen/agent/models.py index b3071fa..1812cd9 100644 --- a/openkyrozen/agent/models.py +++ b/openkyrozen/agent/models.py @@ -19,6 +19,8 @@ class AgentSession: interaction: Any workspace: WorkspaceState messages: list = field(default_factory=list) + history_loaded: bool = False + messages_compacted: bool = False learning: Any = None subagents: Any = None last_learning_run: Any = None diff --git a/openkyrozen/agent/prompts.py b/openkyrozen/agent/prompts.py index 573ab67..bcfcd51 100644 --- a/openkyrozen/agent/prompts.py +++ b/openkyrozen/agent/prompts.py @@ -105,6 +105,42 @@ def _agent_prompt_tools_list(self, agent_config: dict[str, Any]) -> str: return "\n".join(lines) +def _refresh_tool_inventory(self, messages: list[dict[str, Any]]) -> None: + inventory = self._agent_prompt_tools_list(load_agent_config(self._get_workspace_root())) + markers = ( + "## Available read-only tools\n", "## Available tools\n", "## Available Tools\n", + "Available read-only tools:\n", + "Here are the tools you can use. Make sure to pick an action name exactly as listed:\n", + ) + for message in messages: + if not isinstance(message, dict) or message.get("role") != "system": + continue + content = message.get("content") + if not isinstance(content, str): + continue + for marker in markers: + start = content.find(marker) + if start < 0: + continue + body_start = start + len(marker) + end = content.find("\n## ", body_start) + if end < 0: + end = len(content) + message["content"] = content[:body_start] + inventory + content[end:] + break + import hashlib + import json + catalog = [(item.name, item.version, item.input_schema) for item in self.tool_registry.specs.values()] + stable = messages[0].get("content", "") if messages and isinstance(messages[0], dict) else "" + fingerprint = hashlib.sha256(json.dumps({ + "instructions": stable, "catalog": catalog, + "capabilities": sorted(self._permitted_tool_names()), + }, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest() + self._stable_prefix_info.set({ + **self._stable_prefix_info.get(), "fingerprint": fingerprint, + }) + + def _system_prompt(self, tools_list: str, agent_config: dict[str, Any] | None = None) -> str: from openkyrozen.agent.delegation import GUIDANCE tools_list += "\n\n## Automatic sub-agent delegation\n" + GUIDANCE @@ -348,10 +384,6 @@ def _build_messages(self, user_input: str, learned_context: str = "", "content": self._workspace_info() }) - graph_context = self._project_graph_context(user_input) - if graph_context: - messages.append({"role": "system", "content": graph_context}) - project_instructions = format_instructions(self._get_workspace_root()) if project_instructions and self._active_interaction_mode.get() == "agent": messages.append({"role": "system", "content": project_instructions}) @@ -432,13 +464,54 @@ def _build_messages(self, user_input: str, learned_context: str = "", messages.append({"role": "system", "content": learned_context}) history = retain_context_digests(self.short_term_memory, self.SHORT_TERM_CAP * 2) - if not any(is_context_digest(msg) for msg in history): + history_entry_ids = None + native_entries = [entry for entry in self.conversation_history.active_entries() + if entry.kind == "model.response" + and (entry.payload.get("response", {}).get("tool_calls") + or entry.payload.get("response", {}).get("continuation"))] + if native_entries: + continuation = native_entries[-1].payload["response"].get("continuation") or {} + provider = self.execution_context.provider or self.llm_provider + if provider is None: + raise ValueError("Native conversation history requires its original provider") + provider_id = continuation.get("provider_id") + if not hasattr(provider, "_primary"): + from openkyrozen.providers.models import provider_identity + provider_id = provider_identity(provider) + model = continuation.get("request_model", self.DEEPSEEK_MODEL) if hasattr(provider, "_primary") else self.DEEPSEEK_MODEL + history = self.conversation_history.provider_messages( + provider_id=provider_id, model=model, + exclude_turn_id=self._conversation_turn_id.get(), + ) + history_entry_ids = history.entry_ids + elif not any(is_context_digest(msg) for msg in history): digest = self._restore_learning_context_digest() if digest: history.insert(0, digest) + history_start = len(messages) for msg in history: messages.append(msg) messages.append({"role": "user", "content": user_input}) + self._context_history_projection_ids.set( + (id(messages), {history_start + index: entry_id for index, entry_id in enumerate(history_entry_ids)}, + (history_start, history_start + len(history_entry_ids))) + if history_entry_ids is not None else None + ) + import hashlib + import json + stable_text = messages[0]["content"] + "\n" + project_instructions + catalog = [(item.name, item.version, item.input_schema) for item in self.tool_registry.specs.values()] + prefix_fingerprint = hashlib.sha256(json.dumps({ + "instructions": stable_text, + "profile": self._prompt_profile(), + "mode": self._active_interaction_mode.get(), + "capabilities": sorted(self._permitted_tool_names(agent_config)), + "catalog": catalog, + }, sort_keys=True, ensure_ascii=False).encode("utf-8")).hexdigest() + self._stable_prefix_info.set({ + "fingerprint": prefix_fingerprint, + "estimated_tokens": max(1, (len(stable_text) + 3) // 4), + }) return messages diff --git a/openkyrozen/agent/runtime.py b/openkyrozen/agent/runtime.py index 4439c2a..dfba188 100644 --- a/openkyrozen/agent/runtime.py +++ b/openkyrozen/agent/runtime.py @@ -64,6 +64,7 @@ class AgentRuntime: _context_scope, context_usage, _store_context_status, + _record_context_history_metrics, _save_learning_context_digest, _restore_learning_context_digest, _load_reported_context_tokens, @@ -133,6 +134,7 @@ class AgentRuntime: _agent_prompt_tools_list, _system_prompt, _build_messages, + _refresh_tool_inventory, ) # agent.session @@ -288,6 +290,14 @@ def __init__(self, memory, tasks, interaction, workspace): self._turn_context = ContextVar("agent_turn", default=None) self._session_context = ContextVar("agent_session", default=None) self._surface_capability_override = ContextVar("agent_surface_capabilities", default=None) + self._conversation_turn_id = ContextVar("conversation_turn_id", default=None) + self._last_model_response = ContextVar("last_model_response", default=None) + self._record_foreground_model = ContextVar("record_foreground_model", default=False) + self._stable_prefix_info = ContextVar("stable_prefix_info", default={}) + self._context_history_metrics = ContextVar("context_history_metrics", default={}) + self._context_history_seen = ContextVar("context_history_seen", default=None) + self._context_history_projection_ids = ContextVar("context_history_projection_ids", default=None) + self._conversation_expected_head = ContextVar("conversation_expected_head", default=None) self._sessions = {} self._workspaces = {workspace.root: workspace} @@ -308,6 +318,12 @@ def _create_session(self, session_id, owner, workspace, interaction_scope): tasks = self.TaskManager(memory.store, workspace_id=interaction_scope, session_id=session_id, user_id=owner) interaction = self.InteractionController(memory.store, user_id=owner, workspace_id=interaction_scope, session_id=session_id) session = AgentSession(session_id, memory, tasks, interaction, workspace) + from openkyrozen.agent.conversation import ConversationHistory + history = ConversationHistory(memory.store, user_id=owner, workspace_id=interaction_scope, + session_id=str(session_id)) + if history.head: + session.messages = history.display_messages() + session.history_loaded = True session.skills = copy.copy(self.skill_registry) session.skills.disabled_builtins = set(self.skill_registry.disabled_builtins) session.preferences = {key: False if isinstance(value, bool) else "" for key, value in self._user_preferences.items()} @@ -353,6 +369,11 @@ def chat(self, session, message, *, clear_tasks=False, profile=None, memory_cont on_event: EventSink | None = None, on_runtime_event: RuntimeEventSink | None = None, approve: Approval | None = None) -> str: with self.use_session(session): + if not getattr(session, "history_loaded", False): + history = self.conversation_history + if history.head: + session.messages = history.display_messages() + session.history_loaded = True from .events import RuntimeEventPublisher memory = self.memory_bank publisher = RuntimeEventPublisher( @@ -363,6 +384,11 @@ def chat(self, session, message, *, clear_tasks=False, profile=None, memory_cont on_legacy_event=on_event, ) runtime_events = self._runtime_event_publisher.set(publisher) + conversation_turn = self._conversation_turn_id.set(f"turn_{uuid.uuid4().hex}") + conversation_head = self._conversation_expected_head.set(None) + self._last_model_response.set(None) + history_seen = self._context_history_seen.set(set()) + history_projection = self._context_history_projection_ids.set(None) events = self._stream_event_callback.set(on_event) native_response = self._last_response_is_native.set(False) turn_failed = self._turn_failed.set(False) @@ -404,6 +430,43 @@ def chat(self, session, message, *, clear_tasks=False, profile=None, memory_cont if task_event is not None: task_event_context.reset(task_event) self._runtime_event_publisher.reset(runtime_events) + self._conversation_turn_id.reset(conversation_turn) + self._conversation_expected_head.reset(conversation_head) + self._context_history_seen.reset(history_seen) + self._context_history_projection_ids.reset(history_projection) + + @property + def conversation_history(self): + from openkyrozen.agent.conversation import ConversationHistory + memory = self.memory_bank + return ConversationHistory( + memory.store, user_id=memory.user_id, + workspace_id=self.interaction_workspace_id(), + session_id=str(self.current_session.session_id), + ) + + def _record_conversation_entry(self, kind, payload, *, call_id=None, task_id=None, request_id=None): + history = self.conversation_history + active_turn = self._conversation_turn_id.get() + parent = self._conversation_expected_head.get() if active_turn else history.head + entry = history.entry( + kind, payload, parent_id=parent, turn_id=self._conversation_turn_id.get(), + request_id=request_id or self._last_model_request_id.get(), + call_id=call_id, task_id=task_id, + ) + history.append(entry, expected_head=parent) + if active_turn: + self._conversation_expected_head.set(entry.id) + return entry + + def reconcile_conversation_messages(self, messages): + """Explicit adapter for callers that still supply a replacement message list.""" + history = self.conversation_history + if history.head is None: + history.import_legacy(messages) + else: + history.reconcile_legacy(messages) + self.short_term_memory = history.display_messages() @property def tool_registry(self): @@ -489,6 +552,7 @@ def short_term_memory(self): @short_term_memory.setter def short_term_memory(self, value): self.current_session.messages = value + self.current_session.messages_compacted = False @property def learning_engine(self): diff --git a/openkyrozen/agent/session.py b/openkyrozen/agent/session.py index 432731d..8bdb39f 100644 --- a/openkyrozen/agent/session.py +++ b/openkyrozen/agent/session.py @@ -95,8 +95,9 @@ def interaction_workspace_id(self, context: LaunchContext | None = None) -> str: def bind_interaction_scope(self, session_id: str, *, user_id: str | None = None) -> None: session = self.open_session(session_id, user_id=user_id) self._default_session = session - current = self.history_manager(session_id).current() - session.messages = (list(current.get("conversation", [])) if current is not None and current.get("kind") != "recovery" else []) + if not session.messages: + current = self.history_manager(session_id).current() + session.messages = (list(current.get("conversation", [])) if current is not None and current.get("kind") != "recovery" else []) self._restore_ponytail_level() diff --git a/openkyrozen/interfaces/cli/chat.py b/openkyrozen/interfaces/cli/chat.py index a1d3fab..25202d0 100644 --- a/openkyrozen/interfaces/cli/chat.py +++ b/openkyrozen/interfaces/cli/chat.py @@ -29,9 +29,7 @@ def _run_cli_chat(self, user_input, interaction_before): self.console.print(f"[{self._ERROR}][Error] Received empty response from LLM[/{self._ERROR}]") return - self.short_term_memory.append({"role": "user", "content": user_input}) cleaned_reply = self._remove_task_blocks(reply) - self.short_term_memory.append({"role": "assistant", "content": cleaned_reply}) self.memory_bank.add_log(f"User: {user_input}\nAssistant: {self._clean_final_response(reply)}") thinking, answer = self._split_reply(reply) @@ -69,8 +67,6 @@ def _run_cli_chat(self, user_input, interaction_before): if len(reply.strip()) < 5: self.console.print(f"[{self._ERROR}][Error] Received empty response from LLM[/{self._ERROR}]") break - self.short_term_memory.append({"role": "user", "content": user_input}) - self.short_term_memory.append({"role": "assistant", "content": reply}) self.memory_bank.add_log(f"User: {user_input}\nAssistant: {self._clean_final_response(reply)}") thinking, answer = self._split_reply(reply) if thinking: diff --git a/openkyrozen/interfaces/cli/rendering.py b/openkyrozen/interfaces/cli/rendering.py index 0b78575..a97a22a 100644 --- a/openkyrozen/interfaces/cli/rendering.py +++ b/openkyrozen/interfaces/cli/rendering.py @@ -244,13 +244,36 @@ def _call_llm_with_spinner(self, messages: list[dict], model: str | None = None, return f"[LLM Error] {exc}" def call_once() -> str: - if streaming: - return self._get_llm_response( - messages, model=model, stream=True, - on_chunk=lambda chunk: self._emit_stream_event({"event": "content", "chunk": str(chunk)}), - on_stream_end=lambda: self._emit_stream_event({"event": "model_complete"}), - ) - return self._get_llm_response(messages, model=model) + record_token = self._record_foreground_model.set( + not private and self._conversation_turn_id.get() is not None + ) + self._last_model_response.set(None) + try: + if streaming: + result = self._get_llm_response( + messages, model=model, stream=True, + on_chunk=lambda chunk: self._emit_stream_event({"event": "content", "chunk": str(chunk)}), + on_stream_end=lambda: self._emit_stream_event({"event": "model_complete"}), + ) + else: + result = self._get_llm_response(messages, model=model) + if not private and self._conversation_turn_id.get() is not None: + if self._last_model_response.get() is None and not self._turn_failed.get(): + from openkyrozen.providers.models import FinishReason, ModelResponse + response = ModelResponse(text=result, finish_reason=FinishReason.UNKNOWN, + metadata={"legacy": True}) + self._last_model_response.set(response) + from openkyrozen.agent.conversation import model_response_to_dict + self._record_conversation_entry("model.response", { + "response": model_response_to_dict(response), + }, request_id=self._last_model_request_id.get()) + elif self._turn_failed.get(): + self._record_conversation_entry("model.failure", { + "error_type": "provider_or_context_failure", + }, request_id=self._last_model_request_id.get()) + return result + finally: + self._record_foreground_model.reset(record_token) def call_with_one_overflow_recovery() -> str: try: diff --git a/openkyrozen/interfaces/tui/turns.py b/openkyrozen/interfaces/tui/turns.py index f3dcc75..b9a035c 100644 --- a/openkyrozen/interfaces/tui/turns.py +++ b/openkyrozen/interfaces/tui/turns.py @@ -301,10 +301,6 @@ def _run_submit(self, text: str, request_id: str) -> None: self.emit("error", request_id, code="empty_response", error="The provider returned no answer.") return self._staged_attachments.clear() - runtime.short_term_memory.extend([ - {"role": "user", "content": text}, - {"role": "assistant", "content": reply}, - ]) self._quiet_call(runtime.memory_bank.add_log, f"User: {text}\nAssistant: {reply}") thinking, answer = runtime._split_reply(reply) if thinking: diff --git a/openkyrozen/interfaces/web/chat.py b/openkyrozen/interfaces/web/chat.py index aa92afa..60d323c 100644 --- a/openkyrozen/interfaces/web/chat.py +++ b/openkyrozen/interfaces/web/chat.py @@ -94,19 +94,24 @@ def _get_or_create_session(self, session_id: str, user_id: str = "anonymous") -> user_id = service._SERVER_ACTOR_ID with service._chat_lock: if session_id not in service._sessions: - persisted = service._agent.memory_bank.store.list_events( - event_type="session.message", limit=service._MAX_SESSION_MESSAGES, - workspace_id=service._agent.memory_bank.workspace_id, session_id=session_id, - user_id=user_id, - ) - messages = [] - for event in reversed(persisted): - payload = event.get("payload", {}) - if payload.get("role") in {"user", "assistant"} and payload.get("content"): - messages.append({"role": payload["role"], "content": payload["content"]}) + runtime = service._agent + bound = runtime.open_session(session_id, user_id=user_id) + with runtime.use_session(bound): + messages = runtime.conversation_history.display_messages() + journal_head = runtime.conversation_history.head + if not messages and journal_head is None: + persisted = runtime.memory_bank.store.list_events( + event_type="session.message", limit=service._MAX_SESSION_MESSAGES, + workspace_id=runtime.memory_bank.workspace_id, session_id=session_id, + user_id=user_id, + ) + for event in reversed(persisted): + payload = event.get("payload", {}) + if payload.get("role") in {"user", "assistant"} and payload.get("content"): + messages.append({"role": payload["role"], "content": payload["content"]}) context_events = service._agent.memory_bank.store.list_events( event_type="context.status", limit=1, - workspace_id=service._agent.memory_bank.workspace_id, session_id=session_id, + workspace_id=runtime.memory_bank.workspace_id, session_id=session_id, user_id=user_id, ) service._sessions[session_id] = { @@ -122,9 +127,20 @@ def _get_or_create_session(self, session_id: str, user_id: str = "anonymous") -> del service._sessions[oldest] session = service._sessions[session_id] service._initialise_session_defaults(session) - current = service._agent.history_manager(session_id).current() - if current is not None and current.get("kind") != "recovery": - session["messages"] = retain_context_digests(list(current.get("conversation", [])), service._MAX_SESSION_MESSAGES) + runtime = service._agent + bound = runtime.open_session(session_id, user_id=session.get("user_id", user_id)) + with runtime.use_session(bound): + history = runtime.conversation_history + if history.head is not None: + session["messages"] = retain_context_digests( + history.display_messages(), service._MAX_SESSION_MESSAGES, + ) + else: + current = runtime.history_manager(session_id).current() + if current is not None and current.get("kind") != "recovery": + session["messages"] = retain_context_digests( + list(current.get("conversation", [])), service._MAX_SESSION_MESSAGES, + ) return session @@ -277,11 +293,17 @@ def _run_session_chat(self, session: dict[str, Any], message: str, *, on_runtime self._initialise_session_defaults(session) runtime = self._agent bound = runtime.open_session(session["session_id"], user_id=self._SERVER_ACTOR_ID) - bound.messages = list(session["messages"]) bound.tasks.recover() bound.last_learning_run = session.get("last_learning_run") bound.learning_notices = [] with runtime.use_session(bound): + history = runtime.conversation_history + legacy_messages = session.get("messages") or [] + if history.head: + bound.messages = history.display_messages() + bound.messages_compacted = False + elif history.head is None and legacy_messages: + bound.messages = list(legacy_messages) recalls = runtime.memory_bank.store.list_events("memory.recalled", limit=1, workspace_id=runtime.memory_bank.workspace_id, session_id=bound.session_id, user_id=self._SERVER_ACTOR_ID) previous_recall_id = recalls[0]["id"] if recalls else None reply = runtime.chat( @@ -291,15 +313,14 @@ def _run_session_chat(self, session: dict[str, Any], message: str, *, on_runtime ) reply = runtime._clean_final_response(reply) session["last_learning_run"] = bound.last_learning_run - bound.messages.extend([{"role": "user", "content": message}, {"role": "assistant", "content": reply}]) - session["messages"] = retain_context_digests(bound.messages, self._MAX_SESSION_MESSAGES) + session["messages"] = retain_context_digests( + runtime.conversation_history.display_messages(), self._MAX_SESSION_MESSAGES, + ) session["interaction"] = runtime.interaction_envelope() session["updated"] = time.time() session["context"] = runtime.context_usage() recalls = runtime.memory_bank.store.list_events("memory.recalled", limit=1, workspace_id=runtime.memory_bank.workspace_id, session_id=bound.session_id, user_id=self._SERVER_ACTOR_ID) session["last_memory_receipt"] = recalls[0]["payload"] if recalls and recalls[0]["id"] != previous_recall_id else None - for role, content in (("user", message), ("assistant", reply)): - runtime.memory_bank.store.append_event("session.message", {"role": role, "content": content}, user_id=self._SERVER_ACTOR_ID, workspace_id=runtime.memory_bank.workspace_id, session_id=bound.session_id) # Single-user web sessions feed the same evidence stream as CLI/TUI. # Attributed or audience-scoped sessions stay out of global learning. if (session.get("speaker") == self._SERVER_ACTOR_ID diff --git a/openkyrozen/persistence/events.py b/openkyrozen/persistence/events.py index 13ea31a..b2618f8 100644 --- a/openkyrozen/persistence/events.py +++ b/openkyrozen/persistence/events.py @@ -21,6 +21,161 @@ def append_event(self, event_type: str, payload: Any, *, user_id: str = "local", ) return event_id + def conversation_head(self, *, user_id: str, workspace_id: str, session_id: str) -> str | None: + with self.database.connection() as db: + row = db.execute( + "SELECT id FROM events WHERE event_type='conversation.entry' AND user_id=? " + "AND workspace_id=? AND session_id=? ORDER BY rowid DESC LIMIT 1", + (user_id, workspace_id, session_id), + ).fetchone() + return row["id"] if row else None + + def append_conversation_entry(self, entry: dict[str, Any], *, expected_head=...): + """Append private journal content atomically; generic event APIs cannot read it.""" + entry_id = entry["id"] + payload = self.database._json(entry) + scope = (entry["user_id"], entry["workspace_id"], entry["session_id"]) + with self.database._lock, self.database.connection() as db: + db.execute("BEGIN IMMEDIATE") + existing = db.execute("SELECT event_type,payload,user_id,workspace_id,session_id FROM events WHERE id=?", + (entry_id,)).fetchone() + if existing: + same_scope = (existing["user_id"], existing["workspace_id"], existing["session_id"]) == scope + if existing["event_type"] == "conversation.entry" and same_scope and existing["payload"] == payload: + db.commit() + return entry_id + db.rollback() + from openkyrozen.agent.conversation import ConversationHistoryError + raise ConversationHistoryError("Conversation entry ID conflicts with an existing entry") + head = db.execute( + "SELECT id FROM events WHERE event_type='conversation.entry' AND user_id=? " + "AND workspace_id=? AND session_id=? ORDER BY rowid DESC LIMIT 1", scope, + ).fetchone() + current_head = head["id"] if head else None + if expected_head is not ... and expected_head != current_head: + db.rollback() + from openkyrozen.agent.conversation import ConversationHistoryError + raise ConversationHistoryError("Conversation history head changed") + if entry.get("parent_id") != current_head: + db.rollback() + from openkyrozen.agent.conversation import ConversationHistoryError + raise ConversationHistoryError("Conversation entry parent is stale") + from .models import utc_now + db.execute( + "INSERT INTO events(id,event_type,payload,user_id,workspace_id,session_id,task_id,created_at) " + "VALUES(?,'conversation.entry',?,?,?,?,?,?)", + (entry_id, payload, *scope, entry.get("task_id"), entry["created_at"] or utc_now()), + ) + db.commit() + return entry_id + + def append_conversation_entries(self, entries: list[dict[str, Any]], *, expected_head=...): + """Append one conversation import as a single head-checked transaction.""" + if not entries: + return () + first = entries[0] + scope = (first["user_id"], first["workspace_id"], first["session_id"]) + from openkyrozen.agent.conversation import ConversationHistoryError + with self.database._lock, self.database.connection() as db: + db.execute("BEGIN IMMEDIATE") + row = db.execute( + "SELECT id FROM events WHERE event_type='conversation.entry' AND user_id=? " + "AND workspace_id=? AND session_id=? ORDER BY rowid DESC LIMIT 1", scope, + ).fetchone() + head = row["id"] if row else None + if expected_head is not ... and expected_head != head: + db.rollback() + raise ConversationHistoryError("Conversation history head changed") + try: + for entry in entries: + if ((entry["user_id"], entry["workspace_id"], entry["session_id"]) != scope + or entry.get("parent_id") != head): + raise ConversationHistoryError("Conversation entry scope or parent is stale") + existing = db.execute("SELECT 1 FROM events WHERE id=?", (entry["id"],)).fetchone() + if existing: + raise ConversationHistoryError("Conversation entry ID conflicts with an existing entry") + db.execute( + "INSERT INTO events(id,event_type,payload,user_id,workspace_id,session_id,task_id,created_at) " + "VALUES(?,'conversation.entry',?,?,?,?,?,?)", + (entry["id"], self.database._json(entry), *scope, entry.get("task_id"), + entry.get("created_at") or utc_now()), + ) + head = entry["id"] + db.commit() + except Exception: + db.rollback() + raise + return tuple(entry["id"] for entry in entries) + + def list_conversation_entries(self, *, user_id: str, workspace_id: str, session_id: str): + with self.database.connection() as db: + rows = db.execute( + "SELECT payload FROM events WHERE event_type='conversation.entry' AND user_id=? " + "AND workspace_id=? AND session_id=? ORDER BY rowid", + (user_id, workspace_id, session_id), + ).fetchall() + return [self.database._loads(row["payload"], {}) for row in rows] + + def conversation_checkpoint(self, node_id: str, *, user_id: str, workspace_id: str, session_id: str): + with self.database.connection() as db: + row = db.execute( + "SELECT payload FROM events WHERE id=? AND event_type='conversation.checkpoint' " + "AND user_id=? AND workspace_id=? AND session_id=?", + (f"checkpoint_{node_id}", user_id, workspace_id, session_id), + ).fetchone() + return self.database._loads(row["payload"], {}) if row else None + + def append_history_rollback(self, entries: list[dict[str, Any]], events: list[dict[str, Any]], *, + user_id: str, workspace_id: str, session_id: str, + expected_head: str | None, checkpoint: dict[str, Any] | None = None) -> None: + """Commit branch selection and its interaction events as one visible state change.""" + scope = (user_id, workspace_id, session_id) + with self.database._lock, self.database.connection() as db: + db.execute("BEGIN IMMEDIATE") + row = db.execute( + "SELECT id FROM events WHERE event_type='conversation.entry' AND user_id=? " + "AND workspace_id=? AND session_id=? ORDER BY rowid DESC LIMIT 1", scope, + ).fetchone() + head = row["id"] if row else None + if head != expected_head: + db.rollback() + from openkyrozen.agent.conversation import ConversationHistoryError + raise ConversationHistoryError("Conversation history head changed") + try: + for entry in entries: + if (entry["user_id"], entry["workspace_id"], entry["session_id"]) != scope or entry.get("parent_id") != head: + from openkyrozen.agent.conversation import ConversationHistoryError + raise ConversationHistoryError("Conversation entry scope or parent is stale") + db.execute( + "INSERT INTO events(id,event_type,payload,user_id,workspace_id,session_id,task_id,created_at) " + "VALUES(?,'conversation.entry',?,?,?,?,?,?)", + (entry["id"], self.database._json(entry), *scope, entry.get("task_id"), + entry.get("created_at") or utc_now()), + ) + head = entry["id"] + if checkpoint is not None: + checkpoint_id = f"checkpoint_{checkpoint['node_id']}" + db.execute( + "INSERT INTO events(id,event_type,payload,user_id,workspace_id,session_id,task_id,created_at) " + "VALUES(?,'conversation.checkpoint',?,?,?,?,?,?) " + "ON CONFLICT(id) DO UPDATE SET payload=excluded.payload " + "WHERE events.event_type='conversation.checkpoint' AND events.user_id=excluded.user_id " + "AND events.workspace_id=excluded.workspace_id AND events.session_id=excluded.session_id", + (checkpoint_id, self.database._json(checkpoint), *scope, None, utc_now()), + ) + for event in events: + db.execute( + "INSERT INTO events(id,event_type,payload,user_id,workspace_id,session_id,task_id,created_at) " + "VALUES(?,?,?,?,?,?,?,?)", + (f"evt_{uuid.uuid4().hex}", event["event_type"], self.database._json(event["payload"]), + *scope, None, utc_now()), + ) + db.commit() + except Exception: + db.rollback() + raise + + def append_plan_auto_accepted(self, payload: dict[str, Any], *, expected_previous: dict[str, Any], user_id: str, workspace_id: str, session_id: str) -> str: """Append a narrowed-plan acceptance only if its accepted predecessor is current.""" @@ -132,7 +287,7 @@ def append_plan_completed(self, payload: dict[str, Any], *, user_id: str, def list_events(self, event_type: str | None = None, *, limit: int | None = 100, workspace_id: str = "default", session_id: str | None = None, user_id: str | None = None) -> list[dict[str, Any]]: - clauses = ["workspace_id=?"] + clauses = ["workspace_id=?", "event_type NOT LIKE 'conversation.%'"] params: list[Any] = [workspace_id] if user_id is not None: clauses.append("user_id=?") diff --git a/openkyrozen/persistence/history.py b/openkyrozen/persistence/history.py index fbc558f..d5eaef9 100644 --- a/openkyrozen/persistence/history.py +++ b/openkyrozen/persistence/history.py @@ -81,6 +81,7 @@ def insert_history_node(self, node: dict[str, Any], *, set_head: bool = False, node.get("workspace_id", "default"), node["session_id"], node["created_at"], ) with self.database._lock, self.database.connection() as db: + db.execute("BEGIN IMMEDIATE") if set_head: current = db.execute( "SELECT head_id FROM history_heads WHERE user_id=? AND workspace_id=? AND session_id=?", @@ -88,6 +89,7 @@ def insert_history_node(self, node: dict[str, Any], *, set_head: bool = False, ).fetchone() current_id = current["head_id"] if current else None if expected_head is not None and current_id != expected_head: + db.rollback() return False db.execute( "INSERT INTO history_nodes(id,parent_id,kind,summary,user_message,assistant_message,conversation," @@ -101,6 +103,18 @@ def insert_history_node(self, node: dict[str, Any], *, set_head: bool = False, (node.get("user_id", "local"), node.get("workspace_id", "default"), node["session_id"], node["id"], node["created_at"]), ) + if "journal_head" in node: + from .models import utc_now + checkpoint_id = f"checkpoint_{node['id']}" + db.execute( + "INSERT INTO events(id,event_type,payload,user_id,workspace_id,session_id,task_id,created_at) " + "VALUES(?,'conversation.checkpoint',?,?,?,?,?,?)", + (checkpoint_id, self.database._json({"node_id": node["id"], + "journal_head": node["journal_head"]}), + node.get("user_id", "local"), node.get("workspace_id", "default"), + node["session_id"], None, utc_now()), + ) + db.commit() return True diff --git a/openkyrozen/persistence/store.py b/openkyrozen/persistence/store.py index bb14b99..8c8c1cb 100644 --- a/openkyrozen/persistence/store.py +++ b/openkyrozen/persistence/store.py @@ -14,6 +14,8 @@ from .catalogs import CatalogsRepository from .deletion import DeletionRepository +_UNSET = object() + class EventStore: def __init__(self, path=None): self.database = SQLiteDatabase(path) @@ -34,6 +36,25 @@ def __init__(self, path=None): def append_event(self, event_type: str, payload: Any, *, user_id: str='local', workspace_id: str='default', session_id: str | None=None, task_id: str | None=None): return self.events.append_event(event_type, payload, user_id=user_id, workspace_id=workspace_id, session_id=session_id, task_id=task_id) + def conversation_head(self, *, user_id: str, workspace_id: str, session_id: str): + return self.events.conversation_head(user_id=user_id, workspace_id=workspace_id, session_id=session_id) + + def append_conversation_entry(self, entry, *, expected_head=...): + return self.events.append_conversation_entry(entry, expected_head=expected_head) + + def append_conversation_entries(self, entries, *, expected_head=...): + return self.events.append_conversation_entries(entries, expected_head=expected_head) + + def append_history_rollback(self, entries, events, **scope): + return self.events.append_history_rollback(entries, events, **scope) + + def list_conversation_entries(self, *, user_id: str, workspace_id: str, session_id: str): + return self.events.list_conversation_entries(user_id=user_id, workspace_id=workspace_id, session_id=session_id) + + def conversation_checkpoint(self, node_id: str, *, user_id: str, workspace_id: str, session_id: str): + return self.events.conversation_checkpoint(node_id, user_id=user_id, workspace_id=workspace_id, + session_id=session_id) + def append_plan_auto_accepted(self, payload: dict[str, Any], *, expected_previous: dict[str, Any], user_id: str, workspace_id: str, session_id: str): return self.events.append_plan_auto_accepted( @@ -77,7 +98,10 @@ def list_history_nodes(self, *, user_id: str='local', workspace_id: str='default def list_history_sessions(self, *, user_id: str='local', workspace_id: str='default', limit: int=100): return self.history.list_history_sessions(user_id=user_id, workspace_id=workspace_id, limit=limit) - def insert_history_node(self, node: dict[str, Any], *, set_head: bool=False, expected_head: str | None=None): + def insert_history_node(self, node: dict[str, Any], *, set_head: bool=False, expected_head: str | None=None, + journal_head=_UNSET): + if journal_head is not _UNSET: + node = {**node, "journal_head": journal_head} return self.history.insert_history_node(node, set_head=set_head, expected_head=expected_head) def set_history_head(self, node_id: str, *, user_id: str='local', workspace_id: str='default', session_id: str, expected_head: str | None=None): diff --git a/openkyrozen/providers/calls.py b/openkyrozen/providers/calls.py index d22b075..7cc2134 100644 --- a/openkyrozen/providers/calls.py +++ b/openkyrozen/providers/calls.py @@ -96,8 +96,11 @@ def _get_model_response(self, messages: list[dict], model: str | None = None, *, if registry.get_spec(item["name"]).capability in allowed] self._last_prompt_tokens = self._last_completion_tokens = 0 self._last_response_is_native.set(False) + self._last_model_response.set(None) self._turn_failed.set(False) selected_model = model or self.DEEPSEEK_MODEL + if self._active_context_state.get() is not None and not self._in_context_compaction.get(): + self._record_context_history_metrics(messages) request_id = f"model_{uuid.uuid4().hex}" request_token = self._active_model_request_id.set(request_id) self._last_model_request_id.set(request_id) @@ -162,6 +165,12 @@ def observe(event): "model": response.metadata.get("model", selected_model), }, model_request_id=request_id) self._last_response_is_native.set(bool(tools) and not bool(response.metadata.get("legacy"))) + self._last_model_response.set(response) + if self._record_foreground_model.get() and self._conversation_turn_id.get(): + from openkyrozen.agent.conversation import model_response_to_dict + self._record_conversation_entry("model.response", { + "response": model_response_to_dict(response), + }, request_id=request_id) except Exception as exc: self._turn_failed.set(True) self._publish_runtime_event("model.failed", { diff --git a/openkyrozen/providers/fallback.py b/openkyrozen/providers/fallback.py index 44d5531..fe39627 100644 --- a/openkyrozen/providers/fallback.py +++ b/openkyrozen/providers/fallback.py @@ -96,7 +96,7 @@ def _fallback_response(self, state, messages, model, tools=None): candidates = [self._primary] + self._fallbacks from openkyrozen.providers.models import ModelResponse, provider_identity, ProviderContractError continuation = next((item.continuation for item in reversed(messages) - if isinstance(item, ModelResponse) and item.tool_calls), None) + if isinstance(item, ModelResponse) and item.continuation is not None), None) if continuation is not None: provider = next((candidate for candidate in candidates if provider_identity(candidate) == continuation.provider_id), None) diff --git a/openkyrozen/workspace/history.py b/openkyrozen/workspace/history.py index 04ab802..48dfef1 100644 --- a/openkyrozen/workspace/history.py +++ b/openkyrozen/workspace/history.py @@ -208,7 +208,12 @@ def ensure_root(self, *, conversation: list[dict[str, Any]] | None = None, file_summary={**summary, "changes": {"added": 0, "changed": 0, "deleted": 0, "total": 0}, "manifest_digest": hashlib.sha256(_json(manifest).encode()).hexdigest()}, ) - self.store.insert_history_node(node, set_head=True) + self.store.insert_history_node( + node, set_head=True, + journal_head=self.store.conversation_head( + user_id=self.user_id, workspace_id=self.workspace_id, session_id=self.session_id, + ), + ) return node def begin_turn(self, *, conversation: list[dict[str, Any]], interaction: dict[str, Any], @@ -218,7 +223,7 @@ def begin_turn(self, *, conversation: list[dict[str, Any]], interaction: dict[st def commit_turn(self, token: TurnToken, *, user_message: str, assistant_message: str, conversation: list[dict[str, Any]], interaction: dict[str, Any], - tasks: list[dict[str, Any]]) -> dict[str, Any]: + tasks: list[dict[str, Any]], journal_head: str | None = None) -> dict[str, Any]: with self._lock: parent = self.store.history_node(token.parent_id, user_id=self.user_id, workspace_id=self.workspace_id, session_id=self.session_id) @@ -236,7 +241,8 @@ def commit_turn(self, token: TurnToken, *, user_message: str, assistant_message: file_summary={**summary, "changes": self._changes(parent_manifest, manifest), "manifest_digest": hashlib.sha256(_json(manifest).encode()).hexdigest()}, ) - if not self.store.insert_history_node(node, set_head=True, expected_head=token.parent_id): + if not self.store.insert_history_node(node, set_head=True, expected_head=token.parent_id, + journal_head=journal_head): shutil.rmtree(self._node_path(node_id), ignore_errors=True) raise HistoryError("history head changed while recording the turn") return node @@ -363,7 +369,12 @@ def rollback(self, node_id: str, *, confirm: str, expected_head: str | None = No snapshot_relpath=recovery_snapshot, file_summary={**recovery_summary, "changes": self._changes(self._manifest(current), recovery_manifest)}, ) - self.store.insert_history_node(recovery) + self.store.insert_history_node( + recovery, + journal_head=self.store.conversation_head( + user_id=self.user_id, workspace_id=self.workspace_id, session_id=self.session_id, + ), + ) head_switched = False try: self._restore(target) @@ -373,15 +384,54 @@ def rollback(self, node_id: str, *, confirm: str, expected_head: str | None = No head_switched = True self.store.replace_tasks(target["tasks"], user_id=self.user_id, workspace_id=self.workspace_id, session_id=self.session_id) - self.store.append_event( - "history.rollback", {"target_id": target["id"], "recovery_id": recovery_id}, + target_checkpoint = self.store.conversation_checkpoint( + target["id"], user_id=self.user_id, + workspace_id=self.workspace_id, session_id=self.session_id, + ) + journal_head = self.store.conversation_head( user_id=self.user_id, workspace_id=self.workspace_id, session_id=self.session_id, ) - self.store.append_event( - "history.restored", {"node_id": target["id"], "interaction": target["interaction"]}, + from openkyrozen.agent.conversation import ConversationEntry + imported = [] + target_journal_head = target_checkpoint.get("journal_head") if target_checkpoint else None + backfill_checkpoint = None + if target_journal_head is None and target.get("conversation"): + parent = journal_head + for index, message in enumerate(target["conversation"]): + if (not isinstance(message, dict) or message.get("role") not in {"user", "assistant"} + or not isinstance(message.get("content"), str)): + continue + item = ConversationEntry( + id=f"conv_{uuid.uuid4().hex}", + kind="user.input" if message["role"] == "user" else "legacy.assistant", + payload={"text": message["content"], "provenance": "legacy_history_node", + "node_id": target["id"], "index": index, "legacy_index": len(imported)}, + parent_id=parent, user_id=self.user_id, + workspace_id=self.workspace_id, session_id=self.session_id, + ) + imported.append(item.as_dict()) + parent = item.id + target_journal_head = parent + backfill_checkpoint = {"node_id": target["id"], "journal_head": target_journal_head} + branch = ConversationEntry( + id=f"conv_{uuid.uuid4().hex}", kind="branch.selection", + payload={"target_node_id": target["id"], + "target_journal_head": target_journal_head, + "recovery_node_id": recovery_id}, + parent_id=imported[-1]["id"] if imported else journal_head, user_id=self.user_id, workspace_id=self.workspace_id, session_id=self.session_id, ) - except Exception: + self.store.append_history_rollback( + [*imported, branch.as_dict()], [ + {"event_type": "history.rollback", + "payload": {"target_id": target["id"], "recovery_id": recovery_id}}, + {"event_type": "history.restored", + "payload": {"node_id": target["id"], "interaction": target["interaction"]}}, + ], user_id=self.user_id, workspace_id=self.workspace_id, + session_id=self.session_id, expected_head=journal_head, + checkpoint=backfill_checkpoint, + ) + except Exception as exc: try: self._restore(recovery) if head_switched: @@ -395,6 +445,9 @@ def rollback(self, node_id: str, *, confirm: str, expected_head: str | None = No ) except Exception as recovery_error: raise HistoryError(f"rollback failed and recovery failed: {recovery_error}") from recovery_error + from openkyrozen.agent.conversation import ConversationHistoryError + if isinstance(exc, ConversationHistoryError): + raise HistoryError("conversation changed during rollback; retry from the current head") from exc raise return target, recovery diff --git a/openkyrozen/workspace/history_service.py b/openkyrozen/workspace/history_service.py index 7e75c6c..0afbfee 100644 --- a/openkyrozen/workspace/history_service.py +++ b/openkyrozen/workspace/history_service.py @@ -32,6 +32,9 @@ def restore_history(self, node_id: str, *, confirm: str, expected_head: str | No current_tasks=list(self.tasks.tasks), ) self.bind_interaction_scope(manager.session_id, user_id=self.memory_bank.user_id) + self.short_term_memory = self.conversation_history.display_messages() + self.current_session.history_loaded = True + self.current_session.messages_compacted = False if self._project_graph is not None: try: self._project_graph.refresh_async(callback=None) diff --git a/tests/test_context_compaction.py b/tests/test_context_compaction.py index 666b8e2..893748f 100644 --- a/tests/test_context_compaction.py +++ b/tests/test_context_compaction.py @@ -42,6 +42,49 @@ def _messages(old_count=4, *, chars=1_000): class ContextCompactionTests(unittest.TestCase): + def test_native_history_is_accounted_without_mutation_and_cannot_be_compacted_away(self): + from openkyrozen.providers.models import FinishReason, ModelResponse, ToolCall, ToolReply + + response = ModelResponse("", tool_calls=(ToolCall("c1", "read_file", {"path": "a"}),), + finish_reason=FinishReason.TOOL_REQUEST) + reply = ToolReply("c1", "read_file", {"text": "contents"}) + messages = [{"role": "user", "content": "question"}, response, reply] + state = ContextState("custom", configured_window=100_000, reserve_tokens=100) + result = compact_for_pressure(messages, state, lambda *_: self.fail("native history must not summarize")) + self.assertEqual(result.messages, messages) + self.assertIs(result.messages[1], response) + self.assertIs(result.messages[2], reply) + self.assertGreater(state.status["input_tokens"], 0) + + oversized = [{"role": "user", "content": "question" * 1_000}, response, reply] + state = ContextState("custom", configured_window=100, reserve_tokens=20) + result = compact_for_pressure(oversized, state, lambda *_: self.fail("native history must not summarize")) + self.assertTrue(result.impossible) + self.assertEqual(result.messages, oversized) + self.assertIs(result.messages[1], response) + self.assertIs(result.messages[2], reply) + + def test_compaction_can_summarize_legacy_prefix_before_native_history(self): + from openkyrozen.providers.models import FinishReason, ModelResponse, ToolCall, ToolReply + + response = ModelResponse("", tool_calls=(ToolCall("c1", "read_file", {"path": "a"}),), + finish_reason=FinishReason.TOOL_REQUEST) + reply = ToolReply("c1", "read_file", {"text": "contents"}) + messages = ([{"role": "system", "content": "fixed"}] + + [{"role": "user", "content": f"old-{index}:" + "x" * 2_000} + for index in range(8)] + + [response, reply, {"role": "user", "content": "current"}]) + result = compact_for_pressure( + messages, ContextState("custom", configured_window=4_000, reserve_tokens=50), + lambda _text, limit: "summary " + "y" * min(100, limit), + ) + self.assertTrue(result.compacted) + self.assertFalse(result.impossible) + self.assertIn(response, result.messages) + self.assertIn(reply, result.messages) + self.assertLess(result.messages.index(response), result.messages.index(reply)) + self.assertEqual(result.messages[-1], {"role": "user", "content": "current"}) + def test_batched_summary_retains_distinct_facts_from_later_batches(self): summaries = iter(("FIRST: " + "x" * 80, "LAST: required fix " + "y" * 80)) result = _bounded_summary("A" * 20 + "B" * 20, lambda _text, _limit: next(summaries), 20, 40) diff --git a/tests/test_conversation_history.py b/tests/test_conversation_history.py new file mode 100644 index 0000000..5109642 --- /dev/null +++ b/tests/test_conversation_history.py @@ -0,0 +1,677 @@ +import tempfile +import unittest +from pathlib import Path +import os +from unittest.mock import patch + +from openkyrozen.agent.conversation import ( + ConversationHistory, ConversationHistoryError, model_response_to_dict, +) +from openkyrozen.persistence.store import EventStore +from openkyrozen.providers.models import FinishReason, ModelResponse, ProviderContinuation, ToolCall +from openkyrozen.tools.models import ToolError, ToolResult + + +class ConversationHistoryTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.store = EventStore(Path(self.directory.name) / "events.sqlite3") + self.history = ConversationHistory( + self.store, user_id="alice", workspace_id="project", session_id="session-a", + ) + + def tearDown(self): + self.directory.cleanup() + + def test_append_is_idempotent_and_private_from_generic_event_api(self): + entry = self.history.entry("user.input", {"text": "private prompt"}, entry_id="entry-a") + first = self.history.append(entry, expected_head=None) + second = self.history.append(entry, expected_head=None) + self.assertEqual(first, second) + self.assertEqual([item.id for item in self.history.entries()], ["entry-a"]) + self.assertEqual(self.store.list_events(workspace_id="project", user_id="alice"), []) + + def test_conflicting_duplicate_and_stale_head_fail(self): + entry = self.history.entry("user.input", {"text": "one"}, entry_id="entry-a") + head = self.history.append(entry) + with self.assertRaises(ConversationHistoryError): + self.history.append(self.history.entry("user.input", {"text": "two"}, entry_id="entry-a")) + with self.assertRaises(ConversationHistoryError): + self.history.append(self.history.entry("user.input", {"text": "later"}), expected_head=None) + self.assertEqual(self.history.head, head) + + def test_unsupported_entry_version_and_incomplete_native_group_fail_safely(self): + from openkyrozen.providers.models import ProviderContractError + + with self.assertRaises(ConversationHistoryError): + self.history.entry("user.input", {"text": "old"}, version=2) + response = ModelResponse( + text="", tool_calls=(ToolCall("call-1", "read_file", {"path": "a"}),), + finish_reason=FinishReason.TOOL_REQUEST, + continuation=ProviderContinuation("fixture", "m1", "test", {"opaque": True}), + ) + self.history.append(self.history.entry("model.response", { + "response": model_response_to_dict(response), + })) + with self.assertRaises(ProviderContractError): + self.history.provider_messages(provider_id="fixture", model="m1") + + def test_legacy_action_result_stays_textual_in_native_projection(self): + from openkyrozen.agent.conversation import model_response_to_dict, tool_result_to_dict + + self.history.append(self.history.entry("model.response", { + "response": model_response_to_dict(ModelResponse(text="Action: read_file README.md")), + })) + self.history.append(self.history.entry("tool.result", { + "call_id": "legacy-action-1", "name": "read_file", + "output": tool_result_to_dict(ToolResult(text="README contents")), "is_error": False, + })) + projected = self.history.provider_messages(provider_id="stub", model="m1") + self.assertEqual(projected[-1], { + "role": "user", "content": "Tool result (read_file): README contents", + }) + + def test_entries_are_defensive_and_scope_isolated(self): + payload = {"nested": {"value": 1}} + entry = self.history.entry("user.input", payload, entry_id="entry-a") + payload["nested"]["value"] = 2 + self.history.append(entry) + payload_read = self.history.entries()[0].payload + payload_read["nested"]["value"] = 3 + self.assertEqual(self.history.entries()[0].payload["nested"]["value"], 1) + other = ConversationHistory(self.store, user_id="alice", workspace_id="project", session_id="session-b") + self.assertEqual(other.entries(), ()) + + def test_explicit_legacy_replacement_branches_without_deleting_prior_records(self): + first = self.history.entry("user.input", {"text": "old"}) + self.history.append(first) + self.history.reconcile_legacy([{"role": "user", "content": "replacement"}]) + self.assertIn(first.id, [entry.id for entry in self.history.entries()]) + self.assertEqual(self.history.display_messages(), [{"role": "user", "content": "replacement"}]) + + def test_legacy_replacement_commits_branch_and_entries_atomically(self): + first = self.history.entry("user.input", {"text": "old"}) + self.history.append(first) + original_head = self.history.head + encode = self.store.database._json + count = 0 + + def fail_second_entry(value): + nonlocal count + count += 1 + if count == 2: + raise RuntimeError("simulated serialization failure") + return encode(value) + + with patch.object(self.store.database, "_json", side_effect=fail_second_entry): + with self.assertRaisesRegex(RuntimeError, "serialization failure"): + self.history.reconcile_legacy([ + {"role": "user", "content": "replacement"}, + {"role": "assistant", "content": "answer"}, + ]) + self.assertEqual(self.history.head, original_head) + self.assertEqual(self.history.display_messages(), [{"role": "user", "content": "old"}]) + + def test_context_metrics_drop_compacted_native_history_entries(self): + from types import SimpleNamespace + from openkyrozen.agent.compaction import ContextState + from openkyrozen.agent.context import _prepare_context_for_call, _record_context_history_metrics + from openkyrozen.providers.models import ModelResponse + + first = self.history.entry("model.response", { + "response": model_response_to_dict(ModelResponse(text="first")), + }) + self.history.append(first) + second = self.history.entry("model.response", { + "response": model_response_to_dict(ModelResponse(text="second")), + }) + self.history.append(second) + retained = ModelResponse(text="second") + messages = [ModelResponse(text="first"), retained] + + from contextvars import ContextVar + from types import SimpleNamespace + runtime = SimpleNamespace( + _active_context_state=ContextVar("active_context", default=ContextState("fixture")), + _in_context_compaction=ContextVar("in_compaction", default=False), + _context_history_projection_ids=ContextVar("projection", default=None), + conversation_history=self.history, + _conversation_turn_id=ContextVar("turn_id", default=None), + _context_history_seen=ContextVar("seen", default=set()), + _context_history_metrics=ContextVar("metrics", default={}), + short_term_memory=[], + _store_context_status=lambda _state: None, + _load_reported_context_tokens=lambda _state: None, + _summarize_context_with_chat_model=lambda *_args: None, + ) + runtime._context_history_projection_ids.set((id(messages), {0: first.id, 1: second.id}, (0, 2))) + result = SimpleNamespace(messages=[retained], history_digest=None, compacted=True, impossible=False) + with patch("openkyrozen.agent.context.compact_for_pressure", return_value=result): + _prepare_context_for_call(runtime, messages) + _record_context_history_metrics(runtime, messages) + metrics = runtime._context_history_metrics.get() + self.assertEqual(metrics["new_entry_count"], 1) + self.assertEqual(metrics["duplicate_entry_count"], 0) + + def test_native_provider_projection_round_trips_continuation_and_results(self): + response = ModelResponse( + text="", tool_calls=(ToolCall("call-1", "read_file", {"path": "a|b"}),), + usage={"prompt_tokens": 3}, finish_reason=FinishReason.TOOL_REQUEST, + metadata={"provider": "stub", "model": "m1"}, + continuation=ProviderContinuation("stub", "m1", "fixture", {"opaque": [1, 2]}), + ) + from openkyrozen.agent.conversation import model_response_to_dict, tool_result_to_dict + encoded_response = model_response_to_dict(response) + result = ToolResult(data={"contents": "same text"}, text="same text", + error=ToolError("missing", "not found")) + self.history.append(self.history.entry("model.response", {"response": encoded_response}, call_id="call-1")) + self.history.append(self.history.entry("tool.result", { + "call_id": "call-1", "name": "read_file", "output": tool_result_to_dict(result), "is_error": True, + }, call_id="call-1")) + projected = self.history.provider_messages(provider_id="stub", model="m1") + self.assertEqual(projected[0].continuation.payload, {"opaque": [1, 2]}) + self.assertEqual(projected[1].output["data"]["contents"], "same text") + with self.assertRaises(ValueError): + self.history.provider_messages(provider_id="stub", model="m2") + + def test_textless_failed_outcome_is_not_projected_as_blank_assistant(self): + self.history.append(self.history.entry("user.input", {"text": "question"})) + self.history.append(self.history.entry("turn.outcome", { + "status": "failed", "error_type": "ProviderError", + })) + self.assertEqual(self.history.display_messages(), [{"role": "user", "content": "question"}]) + + def test_native_projection_includes_local_outcomes_without_model_responses(self): + response = ModelResponse( + "prior answer", continuation=ProviderContinuation("fixture", "m1", "fixture", {"opaque": 1}), + ) + self.history.append(self.history.entry("user.input", {"text": "prior"}, turn_id="turn-1")) + self.history.append(self.history.entry("model.response", { + "response": model_response_to_dict(response), + }, turn_id="turn-1")) + self.history.append(self.history.entry("user.input", {"text": "accept plan"}, turn_id="turn-2")) + self.history.append(self.history.entry("turn.outcome", { + "status": "completed", "text": "There is no pending plan to accept.", + }, turn_id="turn-2")) + + projected = self.history.provider_messages(provider_id="fixture", model="m1") + self.assertEqual(projected[-2:], [ + {"role": "user", "content": "accept plan"}, + {"role": "assistant", "content": "There is no pending plan to accept."}, + ]) + + def test_native_projection_keeps_runtime_outcome_when_it_replaces_model_prose(self): + response = ModelResponse("The work is complete.") + self.history.append(self.history.entry("user.input", {"text": "finish"}, turn_id="turn-1")) + self.history.append(self.history.entry("model.response", { + "response": model_response_to_dict(response), + }, turn_id="turn-1")) + self.history.append(self.history.entry("turn.outcome", { + "status": "failed", "text": "The work remains incomplete."}, turn_id="turn-1")) + self.assertEqual(self.history.provider_messages(provider_id="fixture", model="m1")[-1], { + "role": "assistant", "content": "The work remains incomplete.", + }) + + def test_native_projection_omits_duplicate_model_outcome(self): + response = ModelResponse("The work is complete.") + self.history.append(self.history.entry("user.input", {"text": "finish"}, turn_id="turn-1")) + self.history.append(self.history.entry("model.response", { + "response": model_response_to_dict(response), + }, turn_id="turn-1")) + self.history.append(self.history.entry("turn.outcome", { + "status": "completed", "text": response.text}, turn_id="turn-1")) + projected = self.history.provider_messages(provider_id="fixture", model="m1") + assistants = [item for item in projected if isinstance(item, dict) and item.get("role") == "assistant"] + self.assertEqual(assistants, [{"role": "assistant", "content": response.text}]) + + def test_legacy_import_is_atomic_and_can_retry_after_storage_failure(self): + messages = [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": "second"}, + ] + encode = self.store.database._json + calls = 0 + + def fail_second(value): + nonlocal calls + calls += 1 + if calls == 2: + raise RuntimeError("simulated legacy import failure") + return encode(value) + + with patch.object(self.store.database, "_json", side_effect=fail_second): + with self.assertRaisesRegex(RuntimeError, "simulated"): + self.history.import_legacy(messages) + self.assertIsNone(self.history.head) + self.history.import_legacy(messages) + self.assertEqual(self.history.display_messages(), messages) + + def test_journal_only_session_is_discoverable_without_exposing_payload(self): + self.history.append(self.history.entry("user.input", {"text": "private prompt"})) + sessions = self.store.list_sessions(workspace_id="project", user_id="alice") + self.assertEqual([item["session_id"] for item in sessions], ["session-a"]) + self.assertNotIn("payload", sessions[0]) + self.assertEqual(self.store.list_events(workspace_id="project", user_id="alice"), []) + + def test_workspace_checkpoint_links_private_journal_and_rollback_selects_branch(self): + from openkyrozen.workspace.history import HistoryManager + from openkyrozen.workspace.context import source_scope_id + root = Path(self.directory.name) / "workspace" + state = Path(self.directory.name) / "state" + root.mkdir() + state.mkdir() + manager = HistoryManager(self.store, root, state, source_scope_id=source_scope_id(root), + user_id="alice", workspace_id="project", session_id="session-a") + user = self.history.entry("user.input", {"text": "base"}) + self.history.append(user) + baseline = manager.ensure_root() + checkpoint = self.store.conversation_checkpoint(baseline["id"], user_id="alice", + workspace_id="project", session_id="session-a") + self.assertEqual(checkpoint["journal_head"], user.id) + later = self.history.entry("user.input", {"text": "abandoned"}) + self.history.append(later) + node = manager.commit_turn(manager.begin_turn(conversation=[], interaction={}, tasks=[]), + user_message="abandoned", assistant_message="done", conversation=[], + interaction={}, tasks=[], journal_head=later.id) + manager.rollback(baseline["id"], confirm="rollback", expected_head=node["id"]) + self.assertEqual([item.payload["text"] for item in self.history.active_entries() + if item.kind == "user.input"], ["base"]) + self.assertIn(later.id, [item.id for item in self.history.entries()]) + + def test_nested_branch_selections_do_not_resurrect_abandoned_entries(self): + first = self.history.entry("user.input", {"text": "A"}) + self.history.append(first) + abandoned = self.history.entry("user.input", {"text": "B"}) + self.history.append(abandoned) + selection = self.history.entry("branch.selection", {"target_journal_head": first.id}) + self.history.append(selection) + retained = self.history.entry("user.input", {"text": "C"}) + self.history.append(retained) + self.history.append(self.history.entry("user.input", {"text": "D"})) + self.history.append(self.history.entry("branch.selection", {"target_journal_head": retained.id})) + self.assertEqual([item.payload["text"] for item in self.history.active_entries() + if item.kind == "user.input"], ["A", "C"]) + + +class ConversationRuntimeTests(unittest.TestCase): + def test_prompt_assembly_uses_correlated_native_history(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.providers.models import ToolCall, ToolReply + from openkyrozen.tools import ToolAdapters + + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + class Provider: + name = "fixture" + runtime.llm_provider = Provider() + runtime.DEEPSEEK_MODEL = "m1" + history = runtime.conversation_history + from openkyrozen.providers.models import provider_identity + response = ModelResponse( + "", tool_calls=(ToolCall("call-1", "read_file", {"path": "a"}),), + finish_reason=FinishReason.TOOL_REQUEST, + continuation=ProviderContinuation(provider_identity(runtime.llm_provider), "m1", "fixture", {"opaque": 1}), + ) + history.append(history.entry("user.input", {"text": "old request"})) + history.append(history.entry("model.response", { + "response": model_response_to_dict(response), + }, call_id="call-1")) + history.append(history.entry("tool.result", { + "call_id": "call-1", "name": "read_file", "output": { + "data": {"text": "file"}, "text": "file", "error": None, "metadata": {}, + }, "is_error": False, + }, call_id="call-1")) + messages = runtime._build_messages("next request") + self.assertTrue(any(isinstance(item, ModelResponse) for item in messages)) + replies = [item for item in messages if isinstance(item, ToolReply)] + self.assertEqual([(item.call_id, item.output["text"]) for item in replies], [("call-1", "file")]) + self.assertFalse(any(isinstance(item, dict) and item.get("content", "").startswith("Tool result (") + for item in messages)) + finally: + app.close() + + def test_prompt_assembly_preserves_selected_continuation_model_and_pauses_on_mismatch(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.providers.models import ProviderContractError, provider_identity + from openkyrozen.tools import ToolAdapters + + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + + class Provider: + name = "fixture" + _continuation_id = "provider-instance" + + provider = Provider() + runtime.llm_provider = provider + runtime.DEEPSEEK_MODEL = "selected-model" + response = ModelResponse( + "answer", finish_reason=FinishReason.FINAL, + continuation=ProviderContinuation( + provider_identity(provider), "selected-model", "fixture", {"opaque": True}, + ), + ) + runtime.conversation_history.append(runtime.conversation_history.entry("user.input", { + "text": "prior request", + })) + runtime.conversation_history.append(runtime.conversation_history.entry("model.response", { + "response": model_response_to_dict(response), + })) + messages = runtime._build_messages("next request") + self.assertTrue(any(isinstance(item, ModelResponse) for item in messages)) + runtime.DEEPSEEK_MODEL = "changed-model" + with self.assertRaisesRegex(ProviderContractError, "different provider or model"): + runtime._build_messages("request after model change") + runtime.llm_provider = Provider() + with self.assertRaisesRegex(ProviderContractError, "different provider or model"): + runtime._build_messages("request after provider restart") + finally: + app.close() + + def test_provider_error_is_not_journaled_as_a_completed_turn(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.tools import ToolAdapters + + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + + def failed_turn(*_args, **_kwargs): + runtime._turn_failed.set(True) + return "[LLM Error] provider failed" + + with patch.object(runtime, "_chat_turn_impl", side_effect=failed_turn): + runtime.chat(runtime.current_session, "question") + outcome = [entry for entry in runtime.conversation_history.entries() + if entry.kind == "turn.outcome"][-1] + self.assertEqual(outcome.payload["status"], "failed") + finally: + app.close() + + def test_failed_turn_refreshes_display_cache_before_next_turn(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.tools import ToolAdapters + + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + with patch.object(runtime, "_chat_turn_impl", side_effect=RuntimeError("provider unavailable")): + with self.assertRaisesRegex(RuntimeError, "provider unavailable"): + runtime.chat(runtime.current_session, "first question") + with patch.object(runtime, "_chat_turn_impl", return_value="second answer"): + runtime.chat(runtime.current_session, "second question") + entries = runtime.conversation_history.active_entries() + self.assertEqual([item.payload["text"] for item in entries if item.kind == "user.input"][-2:], + ["first question", "second question"]) + self.assertFalse(any(item.kind == "branch.selection" for item in entries)) + finally: + app.close() + + def test_truncated_display_projection_does_not_replace_canonical_history(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.tools import ToolAdapters + + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + runtime.SHORT_TERM_CAP = 1 + with patch.object(runtime, "_chat_turn_impl", side_effect=["answer 1", "answer 2", "answer 3"]): + for question in ("question 1", "question 2", "question 3"): + runtime.chat(runtime.current_session, question) + entries = runtime.conversation_history.active_entries() + self.assertFalse(any(item.kind == "branch.selection" for item in entries)) + self.assertEqual([item.payload["text"] for item in entries if item.kind == "user.input"][-3:], + ["question 1", "question 2", "question 3"]) + finally: + app.close() + + def test_session_restart_restores_private_journal_without_replaying_tools(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.providers.models import FinishReason + from openkyrozen.tools import ToolAdapters + + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + database = root / "state.sqlite3" + first = build_application(surface="cli", memory=build_memory(database, user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = first.runtime + history = runtime.conversation_history + history.append(history.entry("user.input", {"text": "remember me"})) + history.append(history.entry("model.response", {"response": { + "text": "restored answer", "tool_calls": [], "usage": None, + "finish_reason": FinishReason.FINAL.value, + "metadata": {"provider": "fixture", "model": "m1"}, "continuation": None, + }})) + finally: + first.close() + + second = build_application(surface="cli", memory=build_memory(database, user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + restored = second.runtime.open_session("surface:cli") + self.assertEqual(restored.messages, [{"role": "user", "content": "remember me"}]) + projected = second.runtime.conversation_history.provider_messages(provider_id="fixture", model="m1") + self.assertEqual(projected[-1], {"role": "assistant", "content": "restored answer"}) + self.assertFalse(any(item.kind.startswith("tool") for item in + second.runtime.conversation_history.entries())) + finally: + second.close() + + def test_tool_receipt_records_correlated_typed_result_once(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.providers.models import ToolCall + from openkyrozen.tools import ToolAdapters + + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + "KYROZEN_CLI_CAPABILITIES": "full", "KYROZEN_APPROVAL_MODE": "never", + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + turn = runtime._conversation_turn_id.set("turn-receipt") + head = runtime._conversation_expected_head.set(None) + try: + receipt = runtime.execute_call(runtime.current_session, + ToolCall("call-receipt", "calculate", {"expression": "2+2"})) + finally: + runtime._conversation_expected_head.reset(head) + runtime._conversation_turn_id.reset(turn) + results = [item for item in runtime.conversation_history.entries() + if item.kind == "tool.result"] + self.assertEqual(len(results), 1) + self.assertEqual(results[0].call_id, "call-receipt") + self.assertEqual(results[0].payload["output"]["data"], receipt.tool_result.data) + self.assertEqual(results[0].payload["is_error"], not receipt.success) + finally: + app.close() + + def test_repeated_equal_inputs_are_distinct_runtime_turns_and_interfaces_do_not_duplicate(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.tools import ToolAdapters + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + with patch.object(runtime, "_chat_turn_impl", return_value="same answer"): + self.assertEqual(runtime.chat(runtime.current_session, "same request"), "same answer") + self.assertEqual(runtime.chat(runtime.current_session, "same request"), "same answer") + entries = runtime.conversation_history.entries() + user_entries = [item for item in entries if item.kind == "user.input"] + self.assertEqual([item.payload["text"] for item in user_entries[-2:]], ["same request", "same request"]) + self.assertEqual([item.kind for item in entries].count("turn.outcome"), 2) + self.assertEqual(runtime.short_term_memory[-4:], [ + {"role": "user", "content": "same request"}, + {"role": "assistant", "content": "same answer"}, + {"role": "user", "content": "same request"}, + {"role": "assistant", "content": "same answer"}, + ]) + self.assertFalse(any(item["event_type"].startswith("conversation.") for item in + runtime.memory_bank.store.list_events( + workspace_id=runtime.interaction_workspace_id(), user_id="alice"))) + finally: + app.close() + + def test_model_response_is_journaled_before_next_dispatch_and_write_failure_is_terminal(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.providers.models import FinishReason, ModelResponse + from openkyrozen.tools import ToolAdapters + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + class Provider: + name = "fixture" + calls = 0 + def chat_response(self, _messages, model=None): + self.calls += 1 + return ModelResponse("answer", finish_reason=FinishReason.FINAL, + metadata={"provider": self.name, "model": model}) + provider = Provider() + runtime.llm_provider = provider + turn = runtime._conversation_turn_id.set("turn-fixture") + record = runtime._record_foreground_model.set(True) + try: + response = runtime._get_model_response([{"role": "user", "content": "question"}], "fixture-model") + self.assertEqual(response.text, "answer") + saved = [entry for entry in runtime.conversation_history.entries() + if entry.kind == "model.response"] + self.assertEqual(saved[-1].payload["response"]["text"], "answer") + runtime.memory_bank.store.append_conversation_entry = lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("disk full")) + with self.assertRaises(OSError): + runtime._get_model_response([{"role": "user", "content": "question 2"}], "fixture-model") + self.assertEqual(provider.calls, 2) + finally: + runtime._record_foreground_model.reset(record) + runtime._conversation_turn_id.reset(turn) + finally: + app.close() + + def test_context_uses_stable_prefix_metrics_and_does_not_inject_graph_context(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.tools import ToolAdapters + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + history = runtime.conversation_history + history.append(history.entry("user.input", {"text": "old"})) + history.append(history.entry("turn.outcome", {"status": "completed", "text": "old answer"})) + runtime.short_term_memory = history.display_messages() + with patch.object(runtime, "_project_graph_context", side_effect=AssertionError("graph was injected")): + first = runtime._build_messages("fresh one") + fingerprint = runtime._stable_prefix_info.get()["fingerprint"] + runtime._record_context_history_metrics(first) + self.assertEqual(runtime._context_history_metrics.get()["new_entry_count"], 2) + second = runtime._build_messages("fresh two") + self.assertEqual(runtime._stable_prefix_info.get()["fingerprint"], fingerprint) + runtime._record_context_history_metrics(second) + metrics = runtime._context_history_metrics.get() + self.assertEqual(metrics["retransmitted_entry_count"], 2) + self.assertEqual(metrics["duplicate_entry_count"], 0) + duplicate = [{"role": "user", "content": "old"}, {"role": "user", "content": "old"}] + runtime._record_context_history_metrics(duplicate) + self.assertEqual(runtime._context_history_metrics.get()["duplicate_entry_count"], 1) + self.assertEqual(runtime._context_history_metrics.get()["retransmitted_entry_count"], 1) + self.assertEqual(sum(item["role"] == "user" and item["content"] == "fresh one" for item in first), 1) + self.assertEqual(sum(item["role"] == "user" and item["content"] == "fresh two" for item in second), 1) + finally: + app.close() + + def test_context_history_metrics_update_for_each_provider_dispatch(self): + from openkyrozen.app.bootstrap import build_application, build_memory + from openkyrozen.agent.compaction import ContextState + from openkyrozen.providers.models import ModelResponse + from openkyrozen.tools import ToolAdapters + + class Provider: + name = "fixture" + + def chat_response(self, _messages, model=None): + return ModelResponse(text="ok", metadata={"provider": self.name, "model": model}) + + with tempfile.TemporaryDirectory() as temp, patch.dict(os.environ, { + "KYROZEN_DISABLE_VECTOR_INDEX": "1", "KYROZEN_WORKSPACE_ROOT": temp, + }): + root = Path(temp) + app = build_application(surface="cli", memory=build_memory(root / "state.sqlite3", user_id="alice", index=None), + tools=ToolAdapters(root), user_id="alice") + try: + runtime = app.runtime + runtime.llm_provider = Provider() + history = runtime.conversation_history + history.append(history.entry("user.input", {"text": "old"})) + history.append(history.entry("turn.outcome", {"status": "completed", "text": "old answer"})) + context = runtime._active_context_state.set(ContextState("fixture")) + seen = runtime._context_history_seen.set(set()) + try: + first_messages = [ + {"role": "user", "content": "old"}, + {"role": "assistant", "content": "old answer"}, + ] + runtime._get_model_response(first_messages, "fixture") + first = runtime._context_history_metrics.get() + self.assertEqual((first["new_entry_count"], first["retransmitted_entry_count"]), (2, 0)) + history.append(history.entry("user.input", {"text": "new"})) + history.append(history.entry("turn.outcome", {"status": "completed", "text": "new answer"})) + second_messages = first_messages + [ + {"role": "user", "content": "new"}, + {"role": "assistant", "content": "new answer"}, + ] + runtime._get_model_response(second_messages, "fixture") + second = runtime._context_history_metrics.get() + self.assertEqual((second["new_entry_count"], second["retransmitted_entry_count"]), (2, 2)) + finally: + runtime._context_history_seen.reset(seen) + runtime._active_context_state.reset(context) + finally: + app.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_dynamic_tools.py b/tests/test_dynamic_tools.py index 699cf70..9c4e087 100644 --- a/tests/test_dynamic_tools.py +++ b/tests/test_dynamic_tools.py @@ -40,6 +40,18 @@ def test_static_validator_rejects_imports_and_process_access(self): self.assertFalse(validate_tool_source("import os\ndef bad(args):\n return os.system(args)\n", "bad")[0]) self.assertFalse(validate_tool_source("def bad(args):\n return open(args).read()\n", "bad")[0]) + def test_registered_tool_replaces_the_prompt_inventory_and_prefix_fingerprint(self): + messages = [{"role": "system", "content": ( + "Prefix\n## Available tools\nold_tool\n\n## Execution protocol\nKeep this section." + )}] + before = main._stable_prefix_info.get().get("fingerprint") + with patch.object(main, "_agent_prompt_tools_list", return_value="new_tool"): + main._refresh_tool_inventory(messages) + self.assertIn("new_tool", messages[0]["content"]) + self.assertNotIn("old_tool", messages[0]["content"]) + self.assertIn("## Execution protocol\nKeep this section.", messages[0]["content"]) + self.assertNotEqual(main._stable_prefix_info.get()["fingerprint"], before) + def test_chat_observer_registers_and_runs_a_real_dynamic_tool(self): source = ( "DefineTool:\n```python\n" diff --git a/tests/test_history.py b/tests/test_history.py index 742be7a..132462b 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -176,6 +176,80 @@ def fail_after_partial_restore(node): self.assertEqual(self.manager.current()["id"], current["id"]) self.assertEqual((self.root / "file.txt").read_text(encoding="utf-8"), "current") + def test_rollback_to_legacy_node_restores_its_text_snapshot_in_journal(self): + from openkyrozen.agent.conversation import ConversationHistory + + legacy = self.manager.ensure_root(conversation=[ + {"role": "user", "content": "legacy prompt"}, + {"role": "assistant", "content": "legacy answer"}, + ]) + with self.store.database.connection() as db: + db.execute("DELETE FROM events WHERE id=?", (f"checkpoint_{legacy['id']}",)) + db.commit() + history = ConversationHistory(self.store, user_id="alice", workspace_id="project", session_id="conversation") + history.append(history.entry("user.input", {"text": "new prompt"})) + current = self.manager.commit_turn( + self.manager.begin_turn(conversation=[], interaction={}, tasks=[]), + user_message="new prompt", assistant_message="new answer", conversation=[], interaction={}, tasks=[], + journal_head=history.head, + ) + self.manager.rollback(legacy["id"], confirm="rollback", expected_head=current["id"]) + self.assertEqual(history.display_messages(), [ + {"role": "user", "content": "legacy prompt"}, + {"role": "assistant", "content": "legacy answer"}, + ]) + checkpoint = self.store.conversation_checkpoint( + legacy["id"], user_id="alice", workspace_id="project", session_id="conversation", + ) + self.assertIsNotNone(checkpoint["journal_head"]) + later = history.entry("user.input", {"text": "later"}) + history.append(later) + current = self.manager.commit_turn( + self.manager.begin_turn(conversation=[], interaction={}, tasks=[]), + user_message="later", assistant_message="answer", conversation=[], interaction={}, tasks=[], + journal_head=later.id, + ) + self.manager.rollback(legacy["id"], confirm="rollback", expected_head=current["id"]) + self.assertEqual(history.display_messages(), [ + {"role": "user", "content": "legacy prompt"}, + {"role": "assistant", "content": "legacy answer"}, + ]) + self.assertEqual(sum(item.payload.get("provenance") == "legacy_history_node" + for item in history.entries()), 2) + + def test_failed_journal_branch_append_does_not_publish_restore_event(self): + from openkyrozen.agent.conversation import ConversationHistory + + baseline = self.manager.ensure_root(interaction={"pending_question": {"id": "q-old"}}) + history = ConversationHistory(self.store, user_id="alice", workspace_id="project", session_id="conversation") + history.append(history.entry("user.input", {"text": "current"})) + current = self.manager.commit_turn( + self.manager.begin_turn(conversation=[], interaction={}, tasks=[]), + user_message="current", assistant_message="answer", conversation=[], interaction={}, tasks=[], + journal_head=history.head, + ) + with patch.object(self.store, "append_history_rollback", side_effect=OSError("database unavailable"), create=True): + with self.assertRaises(OSError): + self.manager.rollback(baseline["id"], confirm="rollback", expected_head=current["id"]) + restored = self.store.list_events("history.restored", user_id="alice", workspace_id="project", + session_id="conversation") + self.assertEqual(restored, []) + self.assertEqual(self.manager.current()["id"], current["id"]) + + def test_stale_journal_head_is_reported_as_history_error(self): + from openkyrozen.agent.conversation import ConversationHistoryError + + baseline = self.manager.ensure_root() + current = self.manager.commit_turn( + self.manager.begin_turn(conversation=[], interaction={}, tasks=[]), + user_message="current", assistant_message="answer", conversation=[], interaction={}, tasks=[], + ) + with patch.object(self.store, "append_history_rollback", + side_effect=ConversationHistoryError("stale journal head"), create=True): + with self.assertRaises(HistoryError) as raised: + self.manager.rollback(baseline["id"], confirm="rollback", expected_head=current["id"]) + self.assertIsInstance(raised.exception.__cause__, ConversationHistoryError) + def test_invalid_source_scope_cannot_escape_snapshot_root(self): with self.assertRaises(HistoryError): HistoryManager(self.store, self.root, self.state, source_scope_id=str(self.root), session_id="x") diff --git a/tests/test_interaction.py b/tests/test_interaction.py index b977327..75f5879 100644 --- a/tests/test_interaction.py +++ b/tests/test_interaction.py @@ -1328,6 +1328,47 @@ def restore(): self.agent.bind_interaction_scope("chat-empty") self.assertEqual(self.agent.short_term_memory, []) + def test_early_plan_acceptance_reply_is_saved_to_session_and_history(self): + session = self.agent.open_session("early-plan-acceptance") + reply = self.agent.chat(session, "accept plan") + self.assertEqual(reply, "There is no pending plan to accept.") + self.assertEqual(session.messages, [ + {"role": "user", "content": "accept plan"}, + {"role": "assistant", "content": "There is no pending plan to accept."}, + ]) + nodes = self.agent.history_manager(session.session_id).list() + self.assertEqual(nodes[-1]["user_message"], "accept plan") + self.assertEqual(nodes[-1]["assistant_message"], reply) + + def test_restore_history_refreshes_active_session_transcript(self): + from openkyrozen.agent.conversation import ConversationHistory + from types import SimpleNamespace + + session_id = "rollback-cache-refresh" + session = self.agent.open_session(session_id) + self.agent.bind_interaction_scope(session_id) + history = ConversationHistory( + session.memory.store, user_id=session.memory.user_id, + workspace_id=self.agent.interaction_workspace_id(), session_id=session_id, + ) + history.append(history.entry("user.input", {"text": "retained request"})) + history.append(history.entry("turn.outcome", {"status": "completed", "text": "retained answer"})) + session.messages = [{"role": "user", "content": "stale transcript"}] + + class Manager: + def __init__(self, identifier): + self.session_id = identifier + + def current(self): + return {"id": "current"} + + def rollback(self, *_args, **_kwargs): + return {"id": "target"}, {"id": "recovery"} + + with patch.object(self.agent, "history_manager", return_value=Manager(session_id)): + self.agent.restore_history("target", confirm="rollback", expected_head="current") + self.assertEqual(session.messages, history.display_messages()) + def test_question_bounds_and_event_reconstruction(self): question = self.controller.request_question({"questions": [{ "id": "scope", "header": "Scope", "prompt": "Which package?", diff --git a/tests/test_provider_native_tools.py b/tests/test_provider_native_tools.py index e3b46ef..3bdece5 100644 --- a/tests/test_provider_native_tools.py +++ b/tests/test_provider_native_tools.py @@ -184,6 +184,32 @@ def chat_response(self, messages, model=None, *, tools=None): self.assertEqual(primary.calls, ["deepseek-model", "deepseek-model"]) self.assertEqual(fallback.calls, []) + def test_fallback_pins_final_continuation_without_tool_calls(self): + from openkyrozen.providers.base import LLMProvider + from openkyrozen.providers.config import ProviderConfig + from openkyrozen.providers.fallback import FallbackProvider + from openkyrozen.providers.models import with_provider_continuation + + class Probe(LLMProvider): + def __init__(self, provider, response=None): + super().__init__(ProviderConfig(provider=provider, model_simple=f"{provider}-model")) + self.response = response + self.calls = [] + def chat(self, messages, model=None): return "", None + def chat_response(self, messages, model=None, *, tools=None): + self.calls.append(model) + return self.response or ModelResponse(text="reply") + + primary, fallback = Probe("deepseek"), Probe("openai") + first = with_provider_continuation( + ModelResponse(text="fallback answer"), fallback, "openai-model", "probe", {"opaque": True}, + ) + wrapper = FallbackProvider.__new__(FallbackProvider) + wrapper._primary, wrapper._fallbacks = primary, [fallback] + wrapper.chat_response([first, {"role": "user", "content": "next turn"}], "deepseek-model") + self.assertEqual(primary.calls, []) + self.assertEqual(fallback.calls, ["openai-model"]) + def test_native_request_executes_through_runtime_without_action_parsing(self): import tempfile from pathlib import Path diff --git a/tests/test_server.py b/tests/test_server.py index 77dbc9b..6e160a8 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -129,6 +129,100 @@ def test_history_api_lists_tree_rejects_stale_or_unconfirmed_rollbacks_and_resto finally: server._sessions = original_sessions + def test_web_chat_does_not_reconcile_stale_snapshot_over_journal(self): + session_id = f"web-journal-authority-{uuid.uuid4().hex}" + original_sessions = server._sessions + server._sessions = {} + try: + with tempfile.TemporaryDirectory() as directory: + memory = MemoryBank(Path(directory) / "state.sqlite3", user_id=server._SERVER_ACTOR_ID, + workspace_id=server._agent.interaction_workspace_id()) + with patch.object(server._agent.current_session, "memory", memory): + session = server._get_or_create_session(session_id, server._SERVER_ACTOR_ID) + bound = server._agent.open_session(session_id, user_id=server._SERVER_ACTOR_ID) + with server._agent.use_session(bound): + history = server._agent.conversation_history + history.append(history.entry("user.input", {"text": "durable request"})) + history.append(history.entry("turn.outcome", {"status": "completed", "text": "durable answer"})) + session["messages"] = [{"role": "user", "content": "stale snapshot"}] + with patch.object(server._agent, "chat", return_value="next answer"), \ + patch.object(server._agent, "_clean_final_response", side_effect=lambda text: text), \ + patch.object(server._agent, "reconcile_conversation_messages") as reconcile: + server._run_session_chat(session, "next request") + reconcile.assert_not_called() + self.assertEqual(session["messages"][:2], [ + {"role": "user", "content": "durable request"}, + {"role": "assistant", "content": "durable answer"}, + ]) + finally: + for key in list(server._agent._sessions): + if key[2] == session_id: + server._agent._sessions.pop(key, None) + server._sessions = original_sessions + + def test_web_session_read_prefers_journal_over_older_workspace_snapshot(self): + from types import SimpleNamespace + + session_id = f"web-journal-snapshot-{uuid.uuid4().hex}" + original_sessions = server._sessions + server._sessions = {} + try: + with tempfile.TemporaryDirectory() as directory: + memory = MemoryBank(Path(directory) / "state.sqlite3", user_id=server._SERVER_ACTOR_ID, + workspace_id=server._agent.interaction_workspace_id()) + with patch.object(server._agent.current_session, "memory", memory): + bound = server._agent.open_session(session_id, user_id=server._SERVER_ACTOR_ID) + with server._agent.use_session(bound): + history = server._agent.conversation_history + history.append(history.entry("user.input", {"text": "latest journal input"})) + history.append(history.entry("turn.outcome", { + "status": "completed", "text": "latest journal answer", + })) + older_snapshot = {"kind": "chat", "conversation": [ + {"role": "user", "content": "older workspace snapshot"}, + ]} + with patch.object(server._agent, "history_manager", return_value=SimpleNamespace( + current=lambda: older_snapshot, + )): + session = server._get_or_create_session(session_id, server._SERVER_ACTOR_ID) + self.assertEqual(session["messages"], [ + {"role": "user", "content": "latest journal input"}, + {"role": "assistant", "content": "latest journal answer"}, + ]) + finally: + for key in list(server._agent._sessions): + if key[2] == session_id: + server._agent._sessions.pop(key, None) + server._sessions = original_sessions + + def test_empty_canonical_web_branch_does_not_restore_legacy_events(self): + session_id = f"empty-branch-{uuid.uuid4().hex}" + original_sessions = server._sessions + server._sessions = {} + try: + with tempfile.TemporaryDirectory() as directory: + memory = MemoryBank(Path(directory) / "state.sqlite3", user_id=server._SERVER_ACTOR_ID, + workspace_id=server._agent.interaction_workspace_id()) + with patch.object(server._agent.current_session, "memory", memory): + bound = server._agent.open_session(session_id, user_id=server._SERVER_ACTOR_ID) + with server._agent.use_session(bound): + history = server._agent.conversation_history + history.append(history.entry( + "branch.selection", {"target_journal_head": None}, + )) + server._agent.memory_bank.store.append_event( + "session.message", {"role": "user", "content": "abandoned"}, + user_id=server._SERVER_ACTOR_ID, + workspace_id=server._agent.interaction_workspace_id(), session_id=session_id, + ) + session = server._get_or_create_session(session_id, server._SERVER_ACTOR_ID) + self.assertEqual(session["messages"], []) + finally: + for key in list(server._agent._sessions): + if key[2] == session_id: + server._agent._sessions.pop(key, None) + server._sessions = original_sessions + def test_chat_interaction_controls_persist_and_restore_without_new_endpoint(self): client = TestClient(server.app) with tempfile.TemporaryDirectory() as directory: @@ -606,7 +700,8 @@ def fake_chat(message, clear_tasks=False, profile=None, memory_context=None): self.assertEqual(server._run_session_chat(session, "hello"), "reply:hello") self.assertEqual(server._run_session_chat(session, "again"), "reply:again") - self.assertEqual(seen, [[], ["hello", "reply:hello"]]) + # The stub bypasses the runtime turn recorder, so no transcript is synthesized by Web. + self.assertEqual(seen, [[], []]) self.assertIs(server._agent.short_term_memory, original_messages) self.assertIs(server._agent.tasks.tasks, original_tasks) diff --git a/tests/test_streaming.py b/tests/test_streaming.py index f280196..84bbe16 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -288,7 +288,12 @@ def text(item): "session.message", workspace_id="dsml-test", session_id=session_id, ) if event["payload"].get("role") == "assistant" ] - self.assertEqual(assistant_messages[-1]["content"], content) + self.assertEqual(assistant_messages, []) + responses = [entry for entry in memory.store.list_conversation_entries( + user_id=server._SERVER_ACTOR_ID, workspace_id=memory.workspace_id, session_id=session_id, + ) if entry.get("kind") == "model.response"] + self.assertIn("DSML", responses[-1]["payload"]["response"]["text"]) + self.assertNotIn("DSML", content) event_names = [item.get("event") for item in payloads] self.assertLess(event_names.index("usage"), event_names.index("completion")) completion = next(item for item in payloads if item.get("event") == "completion") @@ -342,7 +347,12 @@ def text(item): "session.message", workspace_id="inline-control-test", session_id=session_id, ) if event["payload"].get("role") == "assistant" ] - self.assertEqual(assistant_messages[-1]["content"], main._clean_final_response(content)) + self.assertEqual(assistant_messages, []) + responses = [entry for entry in memory.store.list_conversation_entries( + user_id=server._SERVER_ACTOR_ID, workspace_id=memory.workspace_id, session_id=session_id, + ) if entry.get("kind") == "model.response"] + self.assertIn("Action:", responses[-1]["payload"]["response"]["text"]) + self.assertNotIn("Action:", content) def test_prose_prefixed_alias_controls_are_filtered_before_sse_and_persistence(self): session_id = "stream-prose-alias-regression" @@ -391,7 +401,11 @@ def text(item): "session.message", workspace_id="prose-alias-test", session_id=session_id, ) if event["payload"].get("role") == "assistant" ] - self.assertEqual(assistant_messages[-1]["content"], main._clean_final_response(content)) + self.assertEqual(assistant_messages, []) + responses = [entry for entry in memory.store.list_conversation_entries( + user_id=server._SERVER_ACTOR_ID, workspace_id=memory.workspace_id, session_id=session_id, + ) if entry.get("kind") == "model.response"] + self.assertTrue(responses) def test_tool_name_prose_survives_sse_and_persistence(self): session_id = "stream-tool-name-prose-regression" @@ -438,7 +452,11 @@ def text(item): "session.message", workspace_id="tool-name-prose-test", session_id=session_id, ) if event["payload"].get("role") == "assistant" ] - self.assertEqual(assistant_messages[-1]["content"], expected) + self.assertEqual(assistant_messages, []) + responses = [entry for entry in memory.store.list_conversation_entries( + user_id=server._SERVER_ACTOR_ID, workspace_id=memory.workspace_id, session_id=session_id, + ) if entry.get("kind") == "model.response"] + self.assertEqual(responses[-1]["payload"]["response"]["text"], expected) def test_generic_invoke_tags_split_at_every_boundary_stay_out_of_sse(self): marker = _SplitGenericInvokeProvider.marker @@ -495,7 +513,12 @@ def text(item): "session.message", workspace_id="generic-invoke-test", session_id=session_id, ) if event["payload"].get("role") == "assistant" ] - self.assertEqual(assistant_messages[-1]["content"], expected) + self.assertEqual(assistant_messages, []) + responses = [entry for entry in memory.store.list_conversation_entries( + user_id=server._SERVER_ACTOR_ID, workspace_id=memory.workspace_id, session_id=session_id, + ) if entry.get("kind") == "model.response"] + self.assertIn("invoke name=\"read_file\"", responses[-1]["payload"]["response"]["text"]) + self.assertNotIn("