feat: add TypeSafe reply review plugin - #7386
AhmadIbrahiim wants to merge 4 commits into
Conversation
Reviews an agent's replies against its own system prompt and tool catalog using TypeSafe's System One model (Jev), and appends a correction to the chat context when a reply goes off course, so the next reply self-corrects. Jev answers typed questions with calibrated probabilities rather than generating text, and answers every question in a request in parallel against one ingestion of the state. A five-check review is therefore a single round trip, which is what makes it usable on a live call. The checks read the agent's instructions and tool catalog on every review, so nothing is authored per agent and an agent handoff is picked up automatically. Thresholds are applied in plugin code, not by the model, and are documented as starting points to tune against recorded calls. Two placements: observe (default) runs off the reply path once the assistant's message is committed, so it adds nothing to time-to-first-token. gate is opt-in per check and holds a draft until it clears, redrafting once before any audio, at the cost of the full generation plus one evaluation. A review that cannot run is logged and the turn goes unjudged rather than silencing the agent. Such a result carries evaluated=False, and the public flag is needs_correction rather than ok, so an outage cannot be mistaken for a clean call. This is a review layer, not an enforcement boundary, and the README says so. No new dependencies: the single POST endpoint is called over the agent's shared aiohttp session using the repo's existing error types.
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 7 potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| tools: list[dict[str, Any]] = [] | ||
| for name, tool in llm.ToolContext(agent.tools).function_tools.items(): |
There was a problem hiding this comment.
🟡 Observe mode omits MCP tools
In observe mode, _effective_tools omits activity-created MCP tools and automatic cancellation tools. expected_tool therefore cannot detect missed calls to those available tools.
Learn more
The effective catalog used by generation is AgentActivity.tools. It includes session tools, agent tools, activity-created MCP toolsets, and automatic cancellation helpers. This fallback reconstructs only the first two groups.
Example: An agent configured only with mcp_servers can call lookup_ticket, but an observed reply reports an empty TypeSafe catalog. The expected_tool check is skipped instead of catching a reply that omitted lookup_ticket.
Recommended fix: Read and flatten the active activity’s effective tool catalog through a supported framework API. If no public accessor exists, add one rather than duplicating the catalog assembly rules in the plugin.
Was this helpful? React with 👍 or 👎 to provide feedback.
The shipped defaults were guesses and two of them fired on compliant replies. Measured against Jev 1.13 on a labelled set, a compliant reply scores 0.46 to 0.93 on follows_instructions while a violating one scores 0.01 to 0.03, so the old 0.5 sat inside the compliant band. advances_task had 0.02 of margin and severity had 0.13. Jev is close to certain about a violation and much less certain that nothing is wrong, because spotting one broken rule is easier than confirming every rule held. The thresholds therefore belong near the violating end rather than halfway. follows_instructions 0.5 -> 0.25 advances_task 0.4 -> 0.25 severity 2.0 -> 2.2 unsupported_claim and the expected_tool confidence floor were already placed correctly and are unchanged.
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
Gate built its state from agent.chat_ctx, but the pending user message is inserted there only once the speech is scheduled, so a gated review judged a draft without the request that prompted it. Gate now passes the context and tool list it was handed for the generation, and _build_state takes both explicitly. Gate also evaluated only the gated subset and then marked the reply so the committed item was skipped, which left every non-gated check unevaluated for that reply: enabling one gated check silently disabled the other four. Gate now evaluates every check and lets only the gated ones hold the draft, so a reply can clear the gate and still be nudged for an observe-only check. Tools are read from the session's effective catalog rather than agent.tools, so tools registered on AgentSession are visible to expected_tool. Modality instructions render with modality="audio". An Instructions object can hold the whole prompt in its audio variant, so a bare render() hid rules the voice model was actually given. The reviewed item is excluded from the transcript by id rather than by text, which used to delete any earlier turn that happened to repeat the reply. _as_stream accepts the scalar str, ChatChunk and None results that llm_node is allowed to return, so gate no longer breaks a custom node. TypeSafe error bodies are logged as lk.pii.error, since APIStatusError stringifies a response body that can echo the prompt or transcript back.
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 5 new potential issues.
5 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| async def _observe(self, reply: str, item_id: str) -> None: | ||
| session = self._session | ||
| if session is None: | ||
| return | ||
| agent = session.current_agent | ||
| try: | ||
| verdict = await self.review(agent, reply, exclude_item_id=item_id) | ||
| if verdict.triggered_checks: | ||
| await self._apply_nudge(agent, verdict) |
There was a problem hiding this comment.
🟡 Handoffs misattribute in-flight reviews
During a handoff, _observe can resolve the new agent for the old agent’s reply. A later handoff instead leaves the resulting nudge on the inactive old agent.
Learn more
conversation_item_added identifies the message but not its producing agent. _on_item schedules _observe, so session.current_agent can change before _observe starts or while the network evaluation is running. The verdict and its destination must stay associated with the agent that generated the message.
Example: Agent A commits “That costs $40” and immediately hands off to Agent B. If _observe starts after update_agent(B), TypeSafe judges A’s reply using B’s instructions and tools, then nudges B for A’s behavior.
Recommended fix: Capture the producing agent when the assistant event arrives, or add stable agent ownership to the event/message. Before applying a completed verdict, route it to the active successor context when that is the intended handoff behavior rather than mutating an inactive agent.
Was this helpful? React with 👍 or 👎 to provide feedback.
Pin DEFAULT_MODEL to jev-1.13.0, the version default_checks was measured against. The thresholds only hold for the distribution they were measured on, and jev-latest moves on TypeSafe's schedule, so following the alias would leave them quietly mis-set. CALIBRATED_FOR names that version, and the reviewer warns once if something else answered while the shipped thresholds are in use. Pass model="jev-latest" to track the alias instead. evaluated now requires every active check to come back. A 200 carrying an empty answers map previously produced a verdict with no triggers and evaluated=True, so an unjudged turn was counted as clean, which is exactly what the flag exists to prevent. A turn with no applicable checks is recorded instead of dropped, so it reaches results and on_verdict rather than vanishing. A reply is judged against the agent that produced it, captured when the message commits rather than resolved when the detached review resumes, and a correction is dropped instead of being written into a different agent after a handoff. This is not airtight: conversation_item_added does not name the message's author, so a handoff landing before that event is still attributed to the successor. Noted in the code and the README. Gate verdicts are queued per draft and applied when the matching item commits. A set collapsed two in-flight generations with identical short text so the second was judged and billed twice; the queue counts them, and deferring also keeps a gate correction behind the reply it is about without relying on ChatContext.insert ordering by created_at.
Second round addressed — 5 of 6 fixed, 1 does not reproducePushed in 5721074. Five findings fixed, one respectfully disputed with evidence below. Fixed
Does not reproduce
VerificationEach fix has a test, and I checked the load-bearing ones actually fail without their fix rather than trusting a green run. Reverting the agent capture to a late
|
There was a problem hiding this comment.
Devin Review found 4 new potential issues.
6 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| answers = response.get("answers", {}) | ||
| self._warn_on_model_drift(response.get("model")) | ||
| verdict = Verdict( | ||
| answers=answers, | ||
| reviewed_reply=state.reviewed_reply, | ||
| duration=time.monotonic() - started, | ||
| model=response.get("model"), | ||
| usage=response.get("usage") or {}, |
There was a problem hiding this comment.
🔴 Malformed responses abort gated replies
A malformed 200 response makes _evaluate access unchecked mappings outside its failure boundary. gate then raises instead of failing open, dropping the reply.
Learn more
The network request is inside the fail-open try, but response parsing starts after that block. JSON values such as null, a list, or {"answers": []} are valid JSON and can arrive with HTTP 200. Calling response.get(...) or answers.get(...) then raises. Observe mode catches that later in _observe, but gate mode has no outer recovery and aborts the generation.
Example: TypeSafe returns HTTP 200 with {"answers": []} during a gated turn. The request succeeds, but answers.get(check.id) raises. The caller hears no reply instead of the unchecked draft.
Recommended fix: Validate the top-level response, answers, usage, and expected answer shapes inside the same fail-open boundary as evaluate. Return and record an evaluated=False verdict for invalid payloads.
Was this helpful? React with 👍 or 👎 to provide feedback.
| try: | ||
| triggered = check.triggers_when(answer, state) | ||
| except (KeyError, TypeError) as e: |
There was a problem hiding this comment.
🔴 Check exceptions abort gated replies
A triggers_when callback raising outside two selected exception types escapes _evaluate. gate then aborts the reply instead of failing open.
Learn more
Check is public and accepts an arbitrary triggers_when callback. The reviewer promises failed checks fail open, and the documented log levels include crashed checks. The current handler only recognizes KeyError and TypeError; a ValueError, IndexError, or application exception escapes. Observe mode's detached wrapper logs it, but gate mode propagates it through the agent's llm_node.
Example: A custom score check converts a missing string with float(answer["value"]), raising ValueError. A gated call loses its generated draft rather than releasing it unjudged.
Recommended fix: Catch callback exceptions at the check boundary, log the crash, mark the verdict unevaluated, and continue fail-open. Preserve CancelledError and other process-control exceptions.
Was this helpful? React with 👍 or 👎 to provide feedback.
| def _queue_gate_verdict(self, text: str, verdict: Verdict) -> None: | ||
| """Hand a gate's verdict to the commit path, keyed by the draft text.""" | ||
| queue = self._pending_gate.get(text) | ||
| if queue is None: | ||
| queue = deque() | ||
| self._pending_gate[text] = queue | ||
| queue.append(verdict) | ||
| self._pending_gate.move_to_end(text) | ||
| # A replayed stream that never commits would otherwise leave its entry | ||
| # behind for a later identical reply to consume. | ||
| while len(self._pending_gate) > _PENDING_GATE_LIMIT: | ||
| stale, _ = self._pending_gate.popitem(last=False) | ||
| logger.debug("evicting an uncommitted gate verdict", extra={"chars": len(stale)}) |
There was a problem hiding this comment.
🟡 Stale verdicts misjudge later replies
When a gated draft never commits, _pending_gate retains its text-keyed verdict. A later identical reply consumes it, skips review, and receives stale correction.
Learn more
The queue cannot identify the generation that produced a draft, so text equality stands in for ownership. Cancellation or interruption can prevent replayed output from becoming a conversation item. Its verdict then remains eligible for any later equal text; repeated abandoned drafts with one text also bypass the eight-key eviction limit and grow that deque without bound.
Example: A gated "Okay." draft triggers a check, but the speech is cancelled before commit. A later clean "Okay." for another user request consumes the old verdict, avoids evaluation, and gets the old nudge.
Recommended fix: Associate verdicts with a framework generation or message identifier and remove them when that generation is abandoned. If no identifier is available, add explicit expiry and bound total queued verdicts, not only distinct text keys.
Was this helpful? React with 👍 or 👎 to provide feedback.
| def detach(self) -> None: | ||
| """Stop observing and cancel any check still in flight.""" | ||
| if self._session is None: | ||
| return | ||
| self._session.off("conversation_item_added", self._on_item) | ||
| self._session.off("function_tools_executed", self._on_tools_executed) | ||
| self._session = None | ||
| in_flight = [t for t in self._tasks if not t.done()] | ||
| for task in in_flight: | ||
| task.cancel() |
There was a problem hiding this comment.
🟡 Reattachment leaks prior session state
After detach, reattaching preserves tool results, queued verdicts, and history. The new session can inherit old evidence or consume a stale verdict.
Learn more
detach() makes the reviewer attachable again by setting _session to None, but only cancels tasks and removes listeners. Session-scoped collections remain populated. A new session beginning with an assistant greeting can be reviewed before any user event clears the old tool bookkeeping, and matching gated text can consume an old queued verdict.
Example: Session A calls lookup_order, then detaches with an uncommitted gated "Hello." verdict. Session B attaches the same reviewer and opens with "Hello."; it consumes Session A's verdict and retains Session A in results.
Recommended fix: Either make a reviewer permanently single-session or clear _tools_used, _tool_results, _pending_gate, and _results during detach or attach. Document the chosen lifecycle contract.
Was this helpful? React with 👍 or 👎 to provide feedback.
Summary
Adds
livekit-plugins-typesafe, which reviews an agent's own replies and nudges it back on course when one goes off track.After the agent speaks, the plugin sends that reply, the agent's
instructions, its tool catalog and the recent transcript to TypeSafe's System One model (Jev). Jev does not generate text. It answers typed yes/no, multiple-choice and rating questions with calibrated probabilities, and it answers every question in a request in parallel against one ingestion of the state. Five checks are therefore a single round trip, which is what makes it usable on a live call. Thresholds are applied in plugin code rather than by the model, and when one is crossed the plugin appends a correction to the chat context so the next reply corrects itself.Nothing is authored per agent. The checks read
instructionsand the tool catalog on every review, so an agent handoff is picked up automatically.Two placements. Observe is the default and runs off the reply path once the assistant's message is committed, so it adds nothing to time-to-first-token. Gate is opt-in per check, holds a draft until it clears and redrafts once before any audio, at the cost of the full generation plus one evaluation. Gate needs an
llm_node, so realtime and speech-to-speech models can only be observed.A review that cannot run is logged and the turn goes unjudged rather than silencing the agent. That result carries
evaluated=False, and the public flag isneeds_correctionrather thanok, so an outage cannot be read as a clean call. The README states plainly that this is a review layer and not an enforcement boundary.No new dependencies. The API is a single POST, called over the agent's shared aiohttp session using the existing
APIStatusErrorandAPIConnectionErrortypes, rather than pulling in the vendor SDK and a second HTTP stack.Installation
Set
TYPESAFE_API_KEY. An example agent is atexamples/voice_agents/reviewer.py.Validation
AgentSessionthrough the fake STT/LLM/TTS harness, covering the event subscription, tool bookkeeping, nudge delivery,detach()and behaviour during an outage.uv run pytest --unit: 3546 passed. The one failure,test_google_credentials.py::test_clear_error_when_project_unresolvable, is unrelated and environment-dependent; it asserts no application default credentials are present and fails on any machine whereGOOGLE_APPLICATION_CREDENTIALSis set.ruff format --check: 1071 files clean.ruff check: clean.mypyviascripts/check_types.py: no issues in 688 source files, withlivekit.plugins.typesafeincluded in the checked packages.Also exercised against the live TypeSafe endpoint (Jev 1.13) end to end: a real
AgentSessionproducing a real reply, reviewed by the real API. Round trips came back in 378 to 884ms for a five-check review at roughly 960 input tokens, which is within the latency budget the design assumes. Unit tests still run against a fake client so CI needs no key; happy to add apytest.mark.plugin("typesafe")integration test if you want one in the suite.That live run is also what set the thresholds. The first pass shipped guesses, and two of them fired on compliant replies. Measured on a labelled set, a compliant reply scores 0.46 to 0.93 on
follows_instructionswhile a violating one scores 0.01 to 0.03, so the original 0.5 sat inside the compliant band. Jev is close to certain about a violation and much less certain that nothing is wrong, since spotting one broken rule is easier than confirming every rule held, so the thresholds belong near the violating end rather than halfway.follows_instructionsmoved 0.5 to 0.25,advances_task0.4 to 0.25, andseverity2.0 to 2.2.unsupported_claimand theexpected_toolconfidence floor measured correctly and are unchanged. The sample is nine replies under one prompt, which the README says plainly.Notes for reviewers
expected_toolassumes one tool per turn. An agent that legitimately calls two will show low confidence, and the check only fires above a confidence floor, so it stays quiet instead of firing wrongly.tests/directory rather than inside the plugin, because they use the sharedfake_sessionharness and the--unitcategory marker.