wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
709 lines
30 KiB
Python
709 lines
30 KiB
Python
"""Partner agent runtime — drives the chat agent loop from IM messages.
|
|
|
|
This replaces the deleted TutorBot engine. A partner has NO engine of its
|
|
own: every inbound message becomes one chat turn executed by
|
|
``ChatOrchestrator`` → ``AgenticChatPipeline`` (the exact loop the product
|
|
chat uses), run inside the partner's synthetic user scope so rag / skills /
|
|
notebook tools read the partner workspace natively.
|
|
|
|
Event → IM mapping:
|
|
|
|
* ``RESULT`` (``metadata.response``) → the reply message
|
|
* ``CONTENT`` with ``call_kind=llm_final_response`` → terminator/ask_user
|
|
text (the loop's RESULT is empty for an unresolved ask_user pause — the
|
|
pending question IS the reply, and the user's next IM message simply
|
|
starts the next turn)
|
|
* narration rounds (``call_role=narration``) → optional ``_progress``
|
|
messages (``send_progress`` channel flag)
|
|
* ``TOOL_CALL`` → optional ``_tool_hint``
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import mimetypes
|
|
from pathlib import Path
|
|
from typing import Any, Awaitable, Callable
|
|
import uuid
|
|
|
|
from deeptutor.core.context import Attachment, UnifiedContext
|
|
from deeptutor.core.stream import StreamEvent, StreamEventType
|
|
from deeptutor.multi_user.paths import user_context
|
|
from deeptutor.partners.bus.events import InboundMessage, OutboundMessage
|
|
from deeptutor.partners.bus.queue import MessageBus
|
|
from deeptutor.partners.helpers import detect_image_mime
|
|
from deeptutor.services.partners.commands import PartnerCommandHandler
|
|
from deeptutor.services.partners.scope import partner_user
|
|
from deeptutor.services.partners.sessions import PartnerSessionStore
|
|
from deeptutor.services.partners.workspace import ensure_partner_workspace, read_soul
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
EventCallback = Callable[[StreamEvent], Awaitable[None]]
|
|
|
|
_MAX_IMAGE_BYTES = 8 * 1024 * 1024
|
|
_MAX_MEDIA_BYTES = 10 * 1024 * 1024
|
|
_TOOL_HINT_MAX_CHARS = 120
|
|
|
|
|
|
def _format_tool_hint(tool_name: str, args: Any) -> str:
|
|
"""One-line IM rendering of a tool call: ``⚙ rag(query="…")``."""
|
|
rendered = ""
|
|
if isinstance(args, dict) and args:
|
|
parts = []
|
|
for key, value in args.items():
|
|
if str(key).startswith("_"):
|
|
continue
|
|
text = str(value)
|
|
if len(text) > 40:
|
|
text = text[:37] + "…"
|
|
parts.append(f"{key}={text!r}" if isinstance(value, str) else f"{key}={text}")
|
|
rendered = ", ".join(parts)
|
|
hint = f"⚙ {tool_name}({rendered})"
|
|
if len(hint) > _TOOL_HINT_MAX_CHARS:
|
|
hint = hint[: _TOOL_HINT_MAX_CHARS - 1] + "…"
|
|
return hint
|
|
|
|
|
|
class PartnerRunner:
|
|
"""Consume a partner's inbound bus and answer with the chat agent loop."""
|
|
|
|
def __init__(
|
|
self,
|
|
partner_id: str,
|
|
config: Any,
|
|
bus: MessageBus,
|
|
store: PartnerSessionStore,
|
|
save_config: Callable[[str, Any], None] | None = None,
|
|
) -> None:
|
|
self.partner_id = partner_id
|
|
self.config = config
|
|
self.bus = bus
|
|
self.store = store
|
|
self.save_config = save_config
|
|
self._session_locks: dict[str, asyncio.Lock] = {}
|
|
self._tasks: set[asyncio.Task] = set()
|
|
|
|
# ── inbound loop ──────────────────────────────────────────────
|
|
|
|
async def run(self) -> None:
|
|
"""Long-running consumer: one task per message, serialised per session."""
|
|
try:
|
|
while True:
|
|
msg = await self.bus.consume_inbound()
|
|
task = asyncio.create_task(
|
|
self._handle_inbound(msg),
|
|
name=f"partner:{self.partner_id}:turn",
|
|
)
|
|
self._tasks.add(task)
|
|
task.add_done_callback(self._tasks.discard)
|
|
except asyncio.CancelledError:
|
|
for task in list(self._tasks):
|
|
task.cancel()
|
|
raise
|
|
|
|
async def _handle_inbound(self, msg: InboundMessage) -> None:
|
|
delivery_meta: dict[str, Any] = {}
|
|
try:
|
|
final = await self.process_message(msg, delivery_meta=delivery_meta)
|
|
except Exception as exc:
|
|
logger.exception(
|
|
"Partner %s failed to process message on %s", self.partner_id, msg.channel
|
|
)
|
|
final = f"Sorry, something went wrong while processing your message: {exc}"
|
|
if final:
|
|
await self.bus.publish_outbound(
|
|
OutboundMessage(
|
|
channel=msg.channel,
|
|
chat_id=msg.chat_id,
|
|
content=final,
|
|
metadata=delivery_meta,
|
|
)
|
|
)
|
|
|
|
# ── one turn ──────────────────────────────────────────────────
|
|
|
|
def _lock_for(self, session_key: str) -> asyncio.Lock:
|
|
lock = self._session_locks.get(session_key)
|
|
if lock is None:
|
|
lock = asyncio.Lock()
|
|
self._session_locks[session_key] = lock
|
|
return lock
|
|
|
|
async def process_message(
|
|
self,
|
|
msg: InboundMessage,
|
|
*,
|
|
on_event: EventCallback | None = None,
|
|
delivery_meta: dict[str, Any] | None = None,
|
|
) -> str:
|
|
"""Run one chat turn for *msg* and return the final reply text.
|
|
|
|
*delivery_meta*, when given, is filled with metadata the caller
|
|
should attach to the final outbound message (e.g. ``_streamed``
|
|
when the reply was already delivered live via stream deltas).
|
|
"""
|
|
session_key = msg.session_key
|
|
async with self._lock_for(session_key):
|
|
command = PartnerCommandHandler(
|
|
partner_id=self.partner_id,
|
|
config=self.config,
|
|
store=self.store,
|
|
save_config=self.save_config,
|
|
).dispatch(msg)
|
|
if command is not None:
|
|
return command.content
|
|
|
|
final, turn_events = await self._run_turn(
|
|
msg, on_event=on_event, delivery_meta=delivery_meta
|
|
)
|
|
self.store.append(
|
|
session_key,
|
|
"user",
|
|
msg.content,
|
|
channel=msg.channel,
|
|
sender_id=msg.sender_id,
|
|
attachments=list((msg.metadata or {}).get("_attachment_records") or []),
|
|
)
|
|
if final:
|
|
self.store.append(
|
|
session_key,
|
|
"assistant",
|
|
final,
|
|
channel=msg.channel,
|
|
events=turn_events or None,
|
|
)
|
|
return final
|
|
|
|
async def _run_turn(
|
|
self,
|
|
msg: InboundMessage,
|
|
*,
|
|
on_event: EventCallback | None = None,
|
|
delivery_meta: dict[str, Any] | None = None,
|
|
) -> tuple[str, list[dict[str, Any]]]:
|
|
ensure_partner_workspace(self.partner_id)
|
|
primary = getattr(self.config, "llm_selection", None) or None
|
|
backup = getattr(self.config, "backup_llm_selection", None) or None
|
|
|
|
final_text, errors, events = await self._execute_turn(
|
|
msg, selection=primary, on_event=on_event, delivery_meta=delivery_meta
|
|
)
|
|
if not final_text and errors and backup and backup != primary:
|
|
logger.warning(
|
|
"Partner %s turn failed on primary model (%s); retrying with backup",
|
|
self.partner_id,
|
|
errors[-1][:200],
|
|
)
|
|
if delivery_meta is not None:
|
|
delivery_meta.pop("_streamed", None)
|
|
final_text, errors, events = await self._execute_turn(
|
|
msg, selection=backup, on_event=on_event, delivery_meta=delivery_meta
|
|
)
|
|
|
|
if not final_text and errors:
|
|
final_text = f"Sorry, the turn failed: {errors[-1]}"
|
|
return final_text, events
|
|
|
|
async def _execute_turn(
|
|
self,
|
|
msg: InboundMessage,
|
|
*,
|
|
selection: dict[str, str] | None,
|
|
on_event: EventCallback | None = None,
|
|
delivery_meta: dict[str, Any] | None = None,
|
|
) -> tuple[str, list[str], list[dict[str, Any]]]:
|
|
"""Run one chat turn with *selection* active; returns (final, errors, events).
|
|
|
|
``events`` is the turn's trace (every StreamEvent except done/session,
|
|
as ``to_dict()`` — the exact shape the web socket forwards live), so the
|
|
web chat can rehydrate its collapsible "Done" activity after a refresh.
|
|
|
|
A failed turn is ``("", [error, …], events)`` — the caller decides
|
|
whether a backup model gets a second attempt. Exceptions are folded into
|
|
the error list so the retry policy sees them too.
|
|
|
|
When the inbound message asks for streaming (``_wants_stream``, set
|
|
by channels whose config enables it), every loop round's text is
|
|
published live as ``_stream_delta`` messages keyed by
|
|
``_stream_id = {turn_id}:{call_id}`` — narration rounds freeze into
|
|
their own IM message when they complete, and the finish round
|
|
becomes the reply (the final outbound is then marked ``_streamed``
|
|
so the channel doesn't send it twice).
|
|
"""
|
|
from deeptutor.runtime.orchestrator import ChatOrchestrator
|
|
from deeptutor.services.model_selection.runtime import (
|
|
activate_llm_selection,
|
|
reset_llm_selection,
|
|
)
|
|
|
|
final_text = ""
|
|
terminator_text = ""
|
|
turn_id = ""
|
|
round_buffers: dict[str, list[str]] = {}
|
|
streamed_rounds: dict[str, str] = {} # call_id → accumulated streamed text
|
|
ended_rounds: set[str] = set()
|
|
errors: list[str] = []
|
|
turn_events: list[dict[str, Any]] = []
|
|
|
|
# Turn setup (context assembly + LLM-selection resolution) runs INSIDE
|
|
# the try so a setup failure folds into the error list instead of
|
|
# propagating as an opaque crash. The common one is a missing active
|
|
# LLM model: get_llm_config() raises LLMConfigError (a plain Exception,
|
|
# not RuntimeError), which previously escaped _execute_turn and surfaced
|
|
# as a bare "Internal error" on the web socket — masking the real
|
|
# message and skipping the backup-model retry. Folding it here keeps the
|
|
# actual reason ("No active LLM model is configured…") and lets
|
|
# _run_turn fall back to the backup selection.
|
|
#
|
|
# activate_llm_selection still runs BEFORE the partner scope is entered:
|
|
# the model catalog lives in the admin workspace, and the scoped config
|
|
# rides the same async context into the orchestrator task.
|
|
llm_token = None
|
|
try:
|
|
context = self._build_context(msg)
|
|
turn_id = str(context.metadata.get("turn_id") or "")
|
|
send_progress = self._channel_delivery_flag(msg.channel, "send_progress", default=True)
|
|
send_tool_hints = self._channel_delivery_flag(
|
|
msg.channel, "send_tool_hints", default=True
|
|
)
|
|
is_im = msg.channel != "web"
|
|
# Streaming requires send_progress: narration rounds stream live as
|
|
# they happen, so with progress muted we keep buffered delivery.
|
|
wants_stream = is_im and send_progress and bool(msg.metadata.get("_wants_stream"))
|
|
|
|
_config, llm_token = activate_llm_selection(selection)
|
|
# Everything — rag / skills / notebooks AND memory — resolves to the
|
|
# partner's own synthetic workspace. The partner-only memory tools
|
|
# (partner_read / partner_memorize / partner_search, force-mounted by
|
|
# the pipeline) own the split-memory model: partner_read folds in the
|
|
# owner's shared L3 on top of the partner's own, partner_memorize
|
|
# writes only the partner's own. Chat's read_memory / write_memory
|
|
# are suppressed on partner turns, so no admin memory override is
|
|
# needed here (and the partner can never write the owner's memory).
|
|
with user_context(partner_user(self.partner_id, name=self.config.name)):
|
|
orchestrator = ChatOrchestrator()
|
|
async for event in orchestrator.handle(context):
|
|
if on_event is not None:
|
|
await on_event(event)
|
|
meta = event.metadata or {}
|
|
|
|
# Capture the trace for rehydration — mirror product chat's
|
|
# persisted ``assistant_events`` (everything but done/session).
|
|
if event.type not in (StreamEventType.DONE, StreamEventType.SESSION):
|
|
turn_events.append(event.to_dict())
|
|
|
|
if event.type == StreamEventType.CONTENT:
|
|
call_id = str(meta.get("call_id") or "")
|
|
round_buffers.setdefault(call_id, []).append(event.content or "")
|
|
if meta.get("call_kind") == "llm_final_response":
|
|
terminator_text += event.content or ""
|
|
if wants_stream and event.content:
|
|
streamed_rounds[call_id] = (
|
|
streamed_rounds.get(call_id, "") + event.content
|
|
)
|
|
await self._publish_stream_delta(msg, turn_id, call_id, event.content)
|
|
|
|
elif event.type == StreamEventType.TOOL_CALL:
|
|
if is_im and send_tool_hints and event.content:
|
|
hint = _format_tool_hint(event.content, meta.get("args"))
|
|
await self._publish_hint(msg, hint, tool_hint=True)
|
|
|
|
elif event.type == StreamEventType.PROGRESS:
|
|
if (
|
|
meta.get("trace_kind") == "call_status"
|
|
and meta.get("call_state") == "complete"
|
|
and meta.get("call_role") == "narration"
|
|
):
|
|
call_id = str(meta.get("call_id") or "")
|
|
text = "".join(round_buffers.pop(call_id, [])).strip()
|
|
if call_id in streamed_rounds:
|
|
# Already streamed live — freeze the segment.
|
|
ended_rounds.add(call_id)
|
|
await self._publish_stream_end(msg, turn_id, call_id)
|
|
elif is_im and send_progress and text:
|
|
await self._publish_hint(msg, text, tool_hint=False)
|
|
|
|
elif event.type == StreamEventType.RESULT and event.source == "chat":
|
|
final_text = str(meta.get("response") or "")
|
|
|
|
elif event.type == StreamEventType.ERROR and event.content:
|
|
errors.append(event.content)
|
|
except Exception as exc:
|
|
logger.exception("Partner %s turn crashed", self.partner_id)
|
|
errors.append(f"{type(exc).__name__}: {exc}")
|
|
finally:
|
|
reset_llm_selection(llm_token)
|
|
|
|
if not final_text.strip():
|
|
final_text = terminator_text.strip()
|
|
final_text = final_text.strip()
|
|
|
|
# Close any stream segments still open (the finish round, or partial
|
|
# rounds after a crash) so channels can flush their edit buffers.
|
|
for call_id in streamed_rounds:
|
|
if call_id not in ended_rounds:
|
|
await self._publish_stream_end(msg, turn_id, call_id)
|
|
# The reply is "already delivered" only when the live-streamed
|
|
# text matches what the caller is about to send.
|
|
if (
|
|
delivery_meta is not None
|
|
and final_text
|
|
and streamed_rounds[call_id].strip() == final_text
|
|
):
|
|
delivery_meta["_streamed"] = True
|
|
|
|
return final_text, errors, turn_events
|
|
|
|
# ── context assembly ──────────────────────────────────────────
|
|
|
|
def _build_context(self, msg: InboundMessage) -> UnifiedContext:
|
|
session_key = msg.session_key
|
|
turn_id = f"partner-{self.partner_id}-{uuid.uuid4().hex[:12]}"
|
|
history = self.store.conversation_history(session_key)
|
|
attachments, attachment_records = self._attachments_from_media(msg.media)
|
|
source_manifest, source_index = self._source_manifest_from_records(
|
|
session_key,
|
|
fresh_records=attachment_records,
|
|
)
|
|
msg.metadata["_attachment_records"] = attachment_records
|
|
|
|
# Partner-scope context blocks (soul / skills / KBs) are assembled
|
|
# inside the partner scope so the same service locators the chat
|
|
# turn-runtime uses resolve to the partner workspace.
|
|
with user_context(partner_user(self.partner_id, name=self.config.name)):
|
|
skills_manifest = self._build_skills_manifest()
|
|
kb_names = self._list_kb_names()
|
|
|
|
metadata: dict[str, Any] = {
|
|
"turn_id": turn_id,
|
|
"source": "partner",
|
|
"partner_id": self.partner_id,
|
|
"channel": msg.channel,
|
|
"chat_id": msg.chat_id,
|
|
"sender_id": msg.sender_id,
|
|
"session_key": session_key,
|
|
# Swaps the system prompt's product identity ("You are DeepTutor")
|
|
# for the partner's user-given identity; the Soul does the rest.
|
|
"agent_identity": {
|
|
"name": self.config.name,
|
|
"description": getattr(self.config, "description", "") or "",
|
|
},
|
|
# NOTE: no ``wait_for_user_reply`` — an ask_user pause makes
|
|
# the pending question the turn's reply (IM semantics).
|
|
}
|
|
channel_meta: dict[str, Any] = {}
|
|
for key, value in (msg.metadata or {}).items():
|
|
key_text = str(key)
|
|
if key_text.startswith("_"):
|
|
continue
|
|
try:
|
|
json.dumps(value)
|
|
channel_meta[key_text] = value
|
|
except TypeError:
|
|
channel_meta[key_text] = str(value)
|
|
if channel_meta:
|
|
metadata["channel_metadata"] = channel_meta
|
|
if source_index:
|
|
metadata["source_index"] = source_index
|
|
cron_job_id = str((msg.metadata or {}).get("_cron_job_id") or "").strip()
|
|
if cron_job_id:
|
|
metadata["cron_job_id"] = cron_job_id
|
|
mcp_tools = getattr(self.config, "mcp_tools", None)
|
|
if isinstance(mcp_tools, list):
|
|
metadata["mcp_tools_filter"] = [str(name) for name in mcp_tools]
|
|
|
|
return UnifiedContext(
|
|
session_id=f"partner:{self.partner_id}:{session_key}",
|
|
user_message=msg.content,
|
|
conversation_history=history,
|
|
enabled_tools=self._resolved_enabled_tools(),
|
|
allowed_builtin_tools=self._resolved_builtin_tools(),
|
|
active_capability="chat",
|
|
knowledge_bases=kb_names,
|
|
attachments=attachments,
|
|
language=self._language(),
|
|
persona_context=read_soul(self.partner_id).strip(),
|
|
skills_manifest=skills_manifest,
|
|
source_manifest=source_manifest,
|
|
metadata=metadata,
|
|
)
|
|
|
|
def _resolved_enabled_tools(self) -> list[str]:
|
|
"""The partner's user-toggleable tool whitelist.
|
|
|
|
``None`` in config means "everything the user could toggle on in
|
|
chat" — partners default to fully equipped; an explicit list (or
|
|
``[]``) is the owner's selection.
|
|
"""
|
|
configured = getattr(self.config, "enabled_tools", None)
|
|
if configured is None:
|
|
from deeptutor.agents._shared.tool_composition import default_optional_tools
|
|
|
|
return default_optional_tools()
|
|
return [str(name) for name in configured]
|
|
|
|
def _resolved_builtin_tools(self) -> list[str] | None:
|
|
"""The partner's allowed built-in (auto-mounted) tools.
|
|
|
|
``None`` in config means "no gating" — every built-in mounts under its
|
|
usual context condition, exactly like the product chat (partners
|
|
default to fully equipped). An explicit list (or ``[]``) restricts the
|
|
built-in surface so an owner can deny e.g. memory access to an
|
|
IM-facing partner. Flows to ``UnifiedContext.allowed_builtin_tools``.
|
|
"""
|
|
configured = getattr(self.config, "builtin_tools", None)
|
|
if configured is None:
|
|
return None
|
|
return [str(name) for name in configured]
|
|
|
|
def _build_skills_manifest(self) -> str:
|
|
try:
|
|
from deeptutor.services.skill.service import (
|
|
get_skill_service,
|
|
render_skills_manifest,
|
|
)
|
|
|
|
service = get_skill_service()
|
|
entries = service.summary_entries()
|
|
always_block = service.load_always_for_context()
|
|
return "\n\n".join(
|
|
part for part in (always_block, render_skills_manifest(entries)) if part
|
|
)
|
|
except Exception:
|
|
logger.warning(
|
|
"Failed to build skills manifest for partner %s", self.partner_id, exc_info=True
|
|
)
|
|
return ""
|
|
|
|
def _list_kb_names(self) -> list[str]:
|
|
try:
|
|
from deeptutor.knowledge.manager import KnowledgeBaseManager
|
|
from deeptutor.services.path_service import get_path_service
|
|
|
|
kb_root = get_path_service().get_knowledge_bases_root()
|
|
if not kb_root.is_dir():
|
|
return []
|
|
return KnowledgeBaseManager(base_dir=str(kb_root)).list_knowledge_bases()
|
|
except Exception:
|
|
logger.warning("Failed to list KBs for partner %s", self.partner_id, exc_info=True)
|
|
return []
|
|
|
|
def _language(self) -> str:
|
|
lang = str(getattr(self.config, "language", "") or "").strip().lower()
|
|
return "zh" if lang.startswith("zh") else "en"
|
|
|
|
def _channel_delivery_flag(self, channel_name: str, name: str, *, default: bool) -> bool:
|
|
channels = getattr(self.config, "channels", None) or {}
|
|
if not isinstance(channels, dict):
|
|
return default
|
|
section = channels.get(channel_name)
|
|
if not isinstance(section, dict):
|
|
return default
|
|
value = section.get(name)
|
|
if value is None:
|
|
camel = "sendProgress" if name == "send_progress" else "sendToolHints"
|
|
value = section.get(camel)
|
|
return value if isinstance(value, bool) else default
|
|
|
|
@staticmethod
|
|
def _attachment_id_for_path(path: Path) -> str:
|
|
try:
|
|
seed = str(path.resolve())
|
|
except OSError:
|
|
seed = str(path)
|
|
return hashlib.sha1(seed.encode("utf-8"), usedforsecurity=False).hexdigest()[:12]
|
|
|
|
def _attachments_from_media(self, media: list[str]) -> tuple[list[Attachment], list[dict]]:
|
|
attachments: list[Attachment] = []
|
|
records: list[dict[str, Any]] = []
|
|
document_records: list[dict[str, Any]] = []
|
|
for raw_path in media or []:
|
|
try:
|
|
path = Path(raw_path)
|
|
if not path.is_file():
|
|
continue
|
|
size = path.stat().st_size
|
|
if size > _MAX_MEDIA_BYTES:
|
|
continue
|
|
data = path.read_bytes()
|
|
attachment_id = self._attachment_id_for_path(path)
|
|
mime_type = mimetypes.guess_type(path.name)[0] or ""
|
|
mime = detect_image_mime(data)
|
|
if mime and size <= _MAX_IMAGE_BYTES:
|
|
encoded = base64.b64encode(data).decode("ascii")
|
|
attachments.append(
|
|
Attachment(
|
|
type="image",
|
|
base64=encoded,
|
|
filename=path.name,
|
|
mime_type=mime,
|
|
id=attachment_id,
|
|
)
|
|
)
|
|
records.append(
|
|
{
|
|
"id": attachment_id,
|
|
"type": "image",
|
|
"filename": path.name,
|
|
"mime_type": mime,
|
|
"path": str(path),
|
|
"size": size,
|
|
}
|
|
)
|
|
continue
|
|
|
|
document_records.append(
|
|
{
|
|
"id": attachment_id,
|
|
"type": "pdf" if path.suffix.lower() == ".pdf" else "file",
|
|
"filename": path.name,
|
|
"mime_type": mime_type,
|
|
"base64": base64.b64encode(data).decode("ascii"),
|
|
"path": str(path),
|
|
"size": size,
|
|
}
|
|
)
|
|
except OSError:
|
|
logger.warning("Skipping unreadable media file: %s", raw_path, exc_info=True)
|
|
|
|
if document_records:
|
|
try:
|
|
from deeptutor.utils.document_extractor import extract_documents_from_records
|
|
|
|
_document_texts, updated_records = extract_documents_from_records(document_records)
|
|
except Exception:
|
|
logger.warning(
|
|
"Failed to extract partner media documents for %s",
|
|
self.partner_id,
|
|
exc_info=True,
|
|
)
|
|
updated_records = [
|
|
{**record, "base64": "", "extracted_chars": 0} for record in document_records
|
|
]
|
|
|
|
for record in updated_records:
|
|
cleaned = {k: v for k, v in record.items() if k != "base64"}
|
|
records.append(cleaned)
|
|
if str(cleaned.get("extracted_text", "") or "").strip():
|
|
attachments.append(
|
|
Attachment(
|
|
type=str(cleaned.get("type") or "file"),
|
|
filename=str(cleaned.get("filename") or ""),
|
|
mime_type=str(cleaned.get("mime_type") or ""),
|
|
id=str(cleaned.get("id") or ""),
|
|
extracted_text=str(cleaned.get("extracted_text") or ""),
|
|
)
|
|
)
|
|
|
|
return attachments, records
|
|
|
|
def _source_manifest_from_records(
|
|
self,
|
|
session_key: str,
|
|
*,
|
|
fresh_records: list[dict[str, Any]],
|
|
) -> tuple[str, dict[str, str]]:
|
|
try:
|
|
from deeptutor.services.session.source_inventory import (
|
|
SourceEntry,
|
|
SourceInventory,
|
|
render_manifest,
|
|
)
|
|
except Exception:
|
|
logger.warning("Failed to import source inventory helpers", exc_info=True)
|
|
return "", {}
|
|
|
|
inv = SourceInventory()
|
|
turn_ordinal = 1
|
|
historical_messages = self.store.messages(session_key, limit=200)
|
|
for message in historical_messages:
|
|
if message.get("role") == "user":
|
|
turn_ordinal += 1
|
|
for record in message.get("attachments") or []:
|
|
self._add_attachment_source(
|
|
inv,
|
|
record,
|
|
fresh=False,
|
|
first_seen_turn=turn_ordinal - 1,
|
|
source_entry_cls=SourceEntry,
|
|
)
|
|
|
|
for record in fresh_records:
|
|
self._add_attachment_source(
|
|
inv,
|
|
record,
|
|
fresh=True,
|
|
first_seen_turn=turn_ordinal,
|
|
source_entry_cls=SourceEntry,
|
|
)
|
|
return render_manifest(inv)
|
|
|
|
@staticmethod
|
|
def _add_attachment_source(
|
|
inv: Any,
|
|
record: dict[str, Any],
|
|
*,
|
|
fresh: bool,
|
|
first_seen_turn: int,
|
|
source_entry_cls: Any,
|
|
) -> None:
|
|
if str(record.get("type", "")).lower() == "image":
|
|
return
|
|
mime = str(record.get("mime_type", "") or "").lower()
|
|
if mime.startswith("image/"):
|
|
return
|
|
text = str(record.get("extracted_text", "") or "")
|
|
attachment_id = str(record.get("id", "") or "").strip()
|
|
if not text.strip() or not attachment_id:
|
|
return
|
|
inv.add(
|
|
source_entry_cls(
|
|
sid=f"at-{attachment_id}",
|
|
kind="attachment",
|
|
name=str(record.get("filename") or "Untitled file"),
|
|
full_text=text,
|
|
fresh=fresh,
|
|
first_seen_turn=first_seen_turn,
|
|
)
|
|
)
|
|
|
|
async def _publish_hint(self, msg: InboundMessage, text: str, *, tool_hint: bool) -> None:
|
|
await self.bus.publish_outbound(
|
|
OutboundMessage(
|
|
channel=msg.channel,
|
|
chat_id=msg.chat_id,
|
|
content=text,
|
|
metadata={"_progress": True, "_tool_hint": tool_hint},
|
|
)
|
|
)
|
|
|
|
async def _publish_stream_delta(
|
|
self, msg: InboundMessage, turn_id: str, call_id: str, delta: str
|
|
) -> None:
|
|
await self.bus.publish_outbound(
|
|
OutboundMessage(
|
|
channel=msg.channel,
|
|
chat_id=msg.chat_id,
|
|
content=delta,
|
|
metadata={"_stream_delta": True, "_stream_id": f"{turn_id}:{call_id}"},
|
|
)
|
|
)
|
|
|
|
async def _publish_stream_end(self, msg: InboundMessage, turn_id: str, call_id: str) -> None:
|
|
await self.bus.publish_outbound(
|
|
OutboundMessage(
|
|
channel=msg.channel,
|
|
chat_id=msg.chat_id,
|
|
content="",
|
|
metadata={"_stream_end": True, "_stream_id": f"{turn_id}:{call_id}"},
|
|
)
|
|
)
|
|
|
|
|
|
__all__ = ["PartnerRunner"]
|