mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-06 01:37:43 +02:00
[eric] agents: the prefix mode is a type, so the ratchet's own assignment typechecks
This commit is contained in:
@@ -8,7 +8,7 @@ import json
|
||||
import logging
|
||||
import time
|
||||
import os
|
||||
from typing import Dict, List, Optional, Union
|
||||
from typing import Dict, List, Literal, Optional, Union
|
||||
from typeguard import typechecked
|
||||
|
||||
from backend.apps.agents.core.models import AgentSession
|
||||
@@ -44,17 +44,22 @@ from backend.apps.agents.manager.AgentManagerProtocol import AgentManagerProtoco
|
||||
# ratchet exists because a recap-bearing turn was already refused, and an override that widened it
|
||||
# back would hand the filter the exact request it just declined.
|
||||
PREFIX_NARROWNESS = ("minimal", "summary", "none")
|
||||
# The three modes as a TYPE, not a convention. It returned bare `str`, so assigning the result back
|
||||
# to `session.history_prefix_sent` (a Literal field) was a standing type error nobody could act on,
|
||||
# and a typo'd fourth mode would have been caught only by the ratchet quietly doing nothing.
|
||||
PrefixMode = Literal["minimal", "summary", "none"]
|
||||
|
||||
|
||||
@typechecked
|
||||
def effective_prefix_mode(session: AgentSession) -> str:
|
||||
def effective_prefix_mode(session: AgentSession) -> PrefixMode:
|
||||
"""The persisted mode, narrowed by any one-turn override, consumed on read."""
|
||||
p_mode = session.history_prefix_mode
|
||||
p_once = session.history_prefix_once
|
||||
session.history_prefix_once = None
|
||||
if p_once is None:
|
||||
return p_mode
|
||||
return max(p_mode, p_once, key=PREFIX_NARROWNESS.index)
|
||||
p_widest: PrefixMode = max(p_mode, p_once, key=PREFIX_NARROWNESS.index)
|
||||
return p_widest
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -113,8 +113,10 @@ def test_an_unselected_turn_reads_the_outputs_dir_ONCE():
|
||||
scans once and hands the rows down."""
|
||||
src = open(COMPOSER).read()
|
||||
i = src.index("p_app_ids = list(selected_app_output_ids or [])")
|
||||
block = src[i:src.index("app_ctx = build_selected_app_context", i) + 400]
|
||||
assert block.count("load_all()") == 1, "exactly one scan per turn"
|
||||
block = src[i:src.index("if app_ctx:", i)]
|
||||
# Count CALLS, not prose: the comment explaining the fix names load_all() too.
|
||||
code = "\n".join(ln.split("#")[0] for ln in block.splitlines())
|
||||
assert code.count("load_all()") == 1, f"exactly one scan per turn, found {code.count('load_all()')}"
|
||||
assert "apps_created_by_session(session.parent_session_id or session.id, p_outputs)" in block
|
||||
assert "build_unselected_app_context(p_outputs)" in block
|
||||
|
||||
|
||||
Reference in New Issue
Block a user