Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
258 changes: 232 additions & 26 deletions memory/message_memory.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,6 @@
ResponseExtractor,
)


async def safe_call(
call,
*args,
Expand All @@ -28,6 +27,39 @@ async def safe_call(
)


async def emit_runtime_memory_update(
context,
) -> None:

emitter = getattr(
context,
"emitter",
None,
)
emit = getattr(
emitter,
"emit",
None,
)

await safe_call(
emit,
{
"type": "runtime_memory_update",
"memory": getattr(
context,
"runtime_memory",
"",
),
"updates": getattr(
context,
"runtime_memory_updates",
0,
),
},
)


def build_runtime_memory_system_prompt() -> str:

return (
Expand All @@ -36,19 +68,33 @@ def build_runtime_memory_system_prompt() -> str:
"Do not output JSON.\n"
"Do not use Markdown headings.\n"
"Do not explain your reasoning or the summarization process.\n"
"Write memory as atomic bullet lines, one semantic entity per line.\n"
"Each line should start with a compact semantic label such as topic, "
"user intent, active topic, open reference, pending choice, offered options, "
"preference, decision, pattern, or interrupted response.\n"
"Avoid writing about JIN's role unless the role itself changed or matters. "
"Describe assistant actions neutrally instead.\n"
"Keep memory actionable: write what helps the next answer, not a recap of "
"what happened.\n"
"Do not merge unrelated facts into one sentence. Prefer separate lines "
"over broad phrasing like 'Topic established: X, specifically Y'.\n"
"Finish every bullet line completely. Never leave a line mid-phrase.\n"
"Preserve still-relevant existing memory. Update it instead of replacing it blindly.\n"
"Drop old details only when they are clearly obsolete, duplicated, or no longer useful.\n"
"Decide the summary depth from the signal in the latest turn.\n"
"Use shallow summarization for simple factual, isolated, or low-signal turns: "
"keep only the dry fact, topic, or unresolved reference that could help "
"the next answer.\n"
"keep one or two bullet lines with only the dry fact, topic, or unresolved "
"reference that could help the next answer.\n"
"Use deep summarization for turns that reveal user intent, project direction, "
"preferences, recurring patterns, decisions, emotional tone, or a meaningful "
"shift in the conversation trajectory.\n"
"shift in the conversation trajectory; use three to six bullet lines when "
"the turn carries that much signal.\n"
"If the user asks a follow-up that depends on prior context, preserve the "
"referent clearly enough for the next brain prompt to resolve it.\n"
"If the user switches topic, keep the new topic without forcing it into the "
"previous one; only note a pattern if the sequence itself is meaningful.\n"
"If the assistant response was aborted or incomplete, mark it as incomplete "
"and do not treat it as resolved.\n"
"Do not infer stable user traits from a single turn.\n"
"Do not over-interpret jokes, tests, or casual topic changes.\n"
"Prefer compact continuity over transcript-like detail.\n"
Expand All @@ -73,7 +119,30 @@ def build_runtime_memory_user_prompt(
f"{user_message.strip()}\n\n"
"Latest JIN answer:\n"
f"{assistant_message.strip()}\n\n"
"Rewrite the runtime memory now."
"Rewrite the runtime memory now as atomic bullet lines."
)


def build_interrupted_assistant_message(
*,
user_message: str,
assistant_message: str,
) -> str:

partial_text = assistant_message.strip()

if not partial_text:
partial_text = (
"No complete assistant answer was delivered."
)

return (
"Assistant response was interrupted by the user and is incomplete. "
"Do not treat this turn as resolved.\n\n"
"Interrupted user topic/request:\n"
f"{user_message.strip()}\n\n"
"Partial assistant text before interruption:\n"
f"{partial_text}"
)


Expand All @@ -93,6 +162,97 @@ def extract_runtime_memory_text(
return text.strip()


def is_runtime_memory_response_truncated(
response: dict,
) -> bool:

finish_reason = (
ResponseExtractor
.extract_finish_reason(
response
)
.lower()
)

return finish_reason in (
"length",
"max_tokens",
)


def looks_like_incomplete_runtime_memory(
text: str,
) -> bool:

stripped = (
text
or ""
).strip()

if not stripped:
return True

if stripped[-1] in (
",",
":",
"(",
"[",
"{",
):
return True

pairs = (
(
"(",
")",
),
(
"[",
"]",
),
(
"{",
"}",
),
)

return any(
stripped.count(open_char)
> stripped.count(close_char)
for open_char, close_char
in pairs
)


async def ask_runtime_memory_model(
*,
service_client,
current_memory: str,
user_message: str,
assistant_message: str,
) -> dict:

return await ask_service_model(
client=service_client,
system_prompt=(
build_runtime_memory_system_prompt()
),
user_prompt=(
build_runtime_memory_user_prompt(
current_memory=current_memory,
user_message=user_message,
assistant_message=assistant_message,
)
),
temperature=(
config.SERVICE_TEMPERATURE
),
max_tokens=(
config.SERVICE_MAX_TOKENS
),
)


async def summarize_runtime_memory(
*,
context,
Expand Down Expand Up @@ -132,33 +292,43 @@ async def summarize_runtime_memory(
)

try:
response = await ask_service_model(
client=service_client,
system_prompt=(
build_runtime_memory_system_prompt()
),
user_prompt=(
build_runtime_memory_user_prompt(
current_memory=current_memory,
user_message=user_message,
assistant_message=assistant_message,
)
),
temperature=(
config.SERVICE_TEMPERATURE
),
max_tokens=(
min(
config.SERVICE_MAX_TOKENS,
512,
)
),
response = await ask_runtime_memory_model(
service_client=service_client,
current_memory=current_memory,
user_message=user_message,
assistant_message=assistant_message,
)

updated_memory = extract_runtime_memory_text(
response
)

if (
is_runtime_memory_response_truncated(
response
)
or looks_like_incomplete_runtime_memory(
updated_memory
)
):
await safe_call(
getattr(
getattr(
context,
"logger",
None,
),
"log_error",
None,
),
"[MEMORY] runtime memory update skipped",
details=(
"Summarizer returned an incomplete memory update."
),
)

return current_memory

if updated_memory:
context.runtime_memory = updated_memory
context.runtime_memory_updates = (
Expand Down Expand Up @@ -186,6 +356,10 @@ async def summarize_runtime_memory(
"[MEMORY] runtime memory updated",
)

await emit_runtime_memory_update(
context
)

return getattr(
context,
"runtime_memory",
Expand Down Expand Up @@ -258,3 +432,35 @@ def schedule_runtime_memory_update(
)

return task


def schedule_interrupted_runtime_memory_update(
*,
context,
) -> asyncio.Task | None:

user_message = getattr(
context,
"runtime_turn_user_message",
"",
)

assistant_message = (
build_interrupted_assistant_message(
user_message=user_message,
assistant_message=getattr(
context,
"runtime_turn_assistant_response",
"",
),
)
)

if not user_message.strip():
return None

return schedule_runtime_memory_update(
context=context,
user_message=user_message,
assistant_message=assistant_message,
)
6 changes: 6 additions & 0 deletions runtime/runtime_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -53,3 +53,9 @@ class RuntimeContext:
background_tasks: set = field(
default_factory=set
)

runtime_turn_user_message: str = ""

runtime_turn_assistant_response: str = ""

runtime_turn_interrupted: bool = False
19 changes: 19 additions & 0 deletions runtime/runtime_stream.py
Original file line number Diff line number Diff line change
Expand Up @@ -170,6 +170,15 @@ def record_token_usage(self):
),
)

def capture_runtime_turn_response(self):

if not self.is_brain_context():
return

self.context.runtime_turn_assistant_response = (
self.stream.response
)

def build_action_log(
self,
action_event_offset: int,
Expand Down Expand Up @@ -306,20 +315,24 @@ async def run(
)

if not is_valid:
self.capture_runtime_turn_response()

await self.stream.finish(
emit=self.emit_to_chat
)

return None

await self.refresh_token_usage()
self.capture_runtime_turn_response()

await self.stream.finish(
emit=self.emit_to_chat
)

await self.refresh_token_usage()
self.record_token_usage()
self.capture_runtime_turn_response()

log_response = self.stream.response

Expand All @@ -340,6 +353,9 @@ async def run(

except asyncio.CancelledError:

self.context.runtime_turn_interrupted = True
self.capture_runtime_turn_response()

await self.logger.log_runtime(
f"{self.runtime_id} stream cancelled."
)
Expand Down Expand Up @@ -372,6 +388,9 @@ async def run(
httpx.RemoteProtocolError,
):

self.context.runtime_turn_interrupted = True
self.capture_runtime_turn_response()

await self.logger.log_system(
"Generation aborted."
)
Expand Down
Loading