feat: structured content blocks for user messages

ntrp/core/content.py

@@ -0,0 +1,71 @@
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, Field
+
+
+class TextContent(BaseModel):
+    type: Literal["text"] = "text"
+    text: str
+
+
+class ImageContent(BaseModel):
+    type: Literal["image"] = "image"
+    media_type: str
+    data: str
+
+
+class ContextContent(BaseModel):
+    type: Literal["context"] = "context"
+    content_type: str
+    content: str | None = None
+    metadata: dict[str, str] | None = None
+
+
+ContentBlock = Annotated[
+    TextContent | ImageContent | ContextContent,
+    Field(discriminator="type"),
+]
+
+MessageContent = str | list[ContentBlock]
+
+_BLOCK_MAP: dict[str, type[BaseModel]] = {
+    "text": TextContent,
+    "image": ImageContent,
+    "context": ContextContent,
+}
+
+
+def parse_block(raw: dict) -> ContentBlock:
+    cls = _BLOCK_MAP.get(raw.get("type", ""))
+    if cls:
+        return cls.model_validate(raw)
+    return TextContent(text=str(raw))
+
+
+def render_context(ctx: ContextContent | dict) -> str:
+    if isinstance(ctx, dict):
+        ctx = ContextContent.model_validate(ctx)
+    tag = ctx.content_type
+    attrs = ""
+    if ctx.metadata:
+        attrs = " " + " ".join(f'{k}="{v}"' for k, v in ctx.metadata.items())
+    if ctx.content:
+        return f"<{tag}{attrs}>\n{ctx.content}\n</{tag}>"
+    return f"<{tag}{attrs} />"
+
+
+def blocks_to_text(content: str | list | None) -> str:
+    if content is None:
+        return ""
+    if isinstance(content, str):
+        return content
+    parts: list[str] = []
+    for block in content:
+        if isinstance(block, dict):
+            block = parse_block(block)
+        match block:
+            case TextContent():
+                parts.append(block.text)
+            case ContextContent():
+                parts.append(render_context(block))
+    return "\n\n".join(parts)

ntrp/llm/anthropic.py

@@ -4,6 +4,7 @@
 import anthropic
 from pydantic import BaseModel
 
+from ntrp.core.content import render_context
 from ntrp.llm.base import CompletionClient
 from ntrp.llm.models import get_model
 from ntrp.llm.types import (
@@ -202,6 +203,8 @@ def _convert_user_content(self, content: str | list) -> str | list[dict]:
                             "source": {"type": "base64", "media_type": block["media_type"], "data": block["data"]},
                         }
                     )
+                case "context":
+                    result.append({"type": "text", "text": render_context(block)})
         return result
 
     def _convert_assistant(self, msg: dict) -> dict:

ntrp/llm/gemini.py

@@ -7,6 +7,7 @@
 from google.genai import types
 from pydantic import BaseModel
 
+from ntrp.core.content import render_context
 from ntrp.llm.base import CompletionClient, EmbeddingClient
 from ntrp.llm.types import (
     Choice,
@@ -123,6 +124,8 @@ def _convert_user(self, msg: dict) -> types.Content:
                             mime_type=block["media_type"],
                         )
                     )
+                case "context":
+                    parts.append(types.Part(text=render_context(block)))
         return types.Content(role="user", parts=parts or [types.Part(text="")])
 
     def _convert_assistant(self, msg: dict) -> types.Content | None:

ntrp/llm/openai.py

@@ -3,6 +3,7 @@
 import openai
 from pydantic import BaseModel
 
+from ntrp.core.content import render_context
 from ntrp.llm.base import CompletionClient, EmbeddingClient
 from ntrp.llm.types import (
     Choice,
@@ -229,6 +230,8 @@ def _convert_user_content(self, content: list) -> list[dict]:
                             "image_url": {"url": f"data:{block['media_type']};base64,{block['data']}"},
                         }
                     )
+                case "context":
+                    result.append({"type": "text", "text": render_context(block)})
         return result
 
     def _parse_response(self, response, model: str) -> CompletionResponse:

ntrp/llm/utils.py

@@ -1,5 +1,7 @@
 import json
 
+from ntrp.core.content import blocks_to_text as blocks_to_text
+
 
 def parse_args(raw: str | dict) -> dict:
     if isinstance(raw, dict):
@@ -8,17 +10,3 @@ def parse_args(raw: str | dict) -> dict:
         return json.loads(raw)
     except (json.JSONDecodeError, TypeError):
         return {}
-
-
-def blocks_to_text(content: str | list | None) -> str:
-    if content is None:
-        return ""
-    if isinstance(content, str):
-        return content
-    parts = []
-    for block in content:
-        if isinstance(block, dict) and block.get("type") == "text":
-            parts.append(block["text"])
-        elif isinstance(block, str):
-            parts.append(block)
-    return "\n\n".join(parts)

ntrp/server/app.py

@@ -218,15 +218,20 @@ async def chat_message(
         raise HTTPException(status_code=400, detail="session_id required")
 
     images = [img.model_dump() for img in request.images] if request.images else None
+    context = request.context or None
 
     # If agent is already running, queue message for safe injection
     active_run = runtime.run_registry.get_active_run(session_id)
     if active_run:
-        active_run.inject_queue.append({"role": "user", "content": build_user_content(request.message, images)})
+        active_run.inject_queue.append(
+            {"role": "user", "content": build_user_content(request.message, images, context)}
+        )
         return {"run_id": active_run.run_id, "session_id": session_id}
 
     try:
-        ctx = await prepare_chat(runtime, request.message, request.skip_approvals, session_id=session_id, images=images)
+        ctx = await prepare_chat(
+            runtime, request.message, request.skip_approvals, session_id=session_id, images=images, context=context
+        )
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
 

ntrp/server/routers/session.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter, Depends, HTTPException
 
 from ntrp.constants import HISTORY_MESSAGE_LIMIT
-from ntrp.llm.utils import blocks_to_text
+from ntrp.core.content import blocks_to_text
 from ntrp.server.deps import require_session_service
 from ntrp.server.runtime import Runtime, get_runtime
 from ntrp.server.schemas import (
@@ -36,9 +36,16 @@ async def get_session_history(svc: SessionService = Depends(require_session_serv
                 for b in raw_content
                 if isinstance(b, dict) and b.get("type") == "image"
             ]
+            context = [
+                {k: v for k, v in b.items() if k != "type" and v is not None}
+                for b in raw_content
+                if isinstance(b, dict) and b.get("type") == "context"
+            ]
             entry: dict = {"role": role, "content": "\n\n".join(text_parts)}
             if images:
                 entry["images"] = images
+            if context:
+                entry["context"] = context
         else:
             entry = {"role": role, "content": blocks_to_text(raw_content)}
 

ntrp/server/schemas.py

@@ -7,12 +7,13 @@
 
 class ImageBlock(BaseModel):
     media_type: str
-    data: str  # base64-encoded
+    data: str
 
 
 class ChatRequest(BaseModel):
     message: str = Field("", max_length=100_000)
     images: list[ImageBlock] = Field(default_factory=list)
+    context: list[dict] = Field(default_factory=list)
     skip_approvals: bool = False
     session_id: str | None = None
 

ntrp/services/chat.py

@@ -1,12 +1,12 @@
 import asyncio
-import re
 from dataclasses import dataclass
 from datetime import UTC, datetime
 
 from ntrp.channel import Channel
 from ntrp.constants import CONVERSATION_GAP_THRESHOLD
 from ntrp.context.models import SessionData, SessionState
 from ntrp.core.agent import Agent
+from ntrp.core.content import ContextContent, ImageContent, TextContent
 from ntrp.core.factory import AgentConfig, create_agent
 from ntrp.core.prompts import INIT_INSTRUCTION, build_system_blocks
 from ntrp.events.internal import RunCompleted, RunStarted
@@ -78,37 +78,44 @@ async def _resolve_session(runtime: Runtime) -> SessionData:
     return SessionData(runtime.session_service.create(), [])
 
 
-def build_user_content(text: str, images: list[dict] | None = None) -> str | list[dict]:
-    if not images:
+def build_user_content(
+    text: str,
+    images: list[dict] | None = None,
+    context: list[dict] | None = None,
+) -> str | list[dict]:
+    if not images and not context:
         return text
-    blocks: list[dict] = []
+    blocks = []
+    if context:
+        blocks.extend(ContextContent(**ctx).model_dump(exclude_none=True) for ctx in context)
     if text:
-        blocks.append({"type": "text", "text": text})
-    blocks.extend({"type": "image", "media_type": img["media_type"], "data": img["data"]} for img in images)
+        blocks.append(TextContent(text=text).model_dump())
+    if images:
+        blocks.extend(ImageContent(**img).model_dump() for img in images)
     return blocks
 
 
-def _time_gap_note(last_activity: datetime) -> str:
+def _time_gap_note(last_activity: datetime) -> dict | None:
     gap = (datetime.now(UTC) - last_activity).total_seconds()
     if gap < CONVERSATION_GAP_THRESHOLD:
-        return ""
+        return None
     hours = gap / 3600
     if hours < 1:
         elapsed = f"{int(gap / 60)} minutes"
     elif hours < 24:
         elapsed = f"{hours:.1f} hours"
     else:
         elapsed = f"{hours / 24:.1f} days"
-    return f"<time_since_last_message>{elapsed}</time_since_last_message>"
-
-
-_TIME_GAP_RE = re.compile(r"\n*<time_since_last_message>.*?</time_since_last_message>", re.DOTALL)
+    return {"content_type": "time_since_last_message", "content": elapsed}
 
 
-def _strip_time_gaps(messages: list[dict]) -> None:
+def _retain_user_content(messages: list[dict]) -> list[dict]:
+    result = []
     for msg in messages:
-        if msg.get("role") == "user" and isinstance(msg.get("content"), str):
-            msg["content"] = _TIME_GAP_RE.sub("", msg["content"])
+        if msg.get("role") == "user" and isinstance(msg.get("content"), list):
+            msg = {**msg, "content": [b for b in msg["content"] if b.get("type") != "context"]}
+        result.append(msg)
+    return result
 
 
 async def _prepare_messages(
@@ -117,6 +124,7 @@ async def _prepare_messages(
     user_message: str,
     last_activity: datetime | None = None,
     images: list[dict] | None = None,
+    context: list[dict] | None = None,
 ) -> list[dict]:
     memory_context = None
     if runtime.memory:
@@ -137,7 +145,7 @@ async def _prepare_messages(
         use_cache_control=_is_anthropic(runtime.config.chat_model),
     )
 
-    _strip_time_gaps(messages)
+    messages = _retain_user_content(messages)
 
     if not messages:
         messages = [{"role": "system", "content": system_blocks}]
@@ -146,12 +154,13 @@ async def _prepare_messages(
     else:
         messages.insert(0, {"role": "system", "content": system_blocks})
 
+    ctx_blocks = list(context or [])
     if last_activity:
-        gap_note = _time_gap_note(last_activity)
-        if gap_note:
-            user_message = f"{user_message}\n\n{gap_note}"
+        time_gap = _time_gap_note(last_activity)
+        if time_gap:
+            ctx_blocks.append(time_gap)
 
-    messages.append({"role": "user", "content": build_user_content(user_message, images)})
+    messages.append({"role": "user", "content": build_user_content(user_message, images, ctx_blocks or None)})
 
     return messages
 
@@ -162,6 +171,7 @@ async def prepare_chat(
     skip_approvals: bool = False,
     session_id: str | None = None,
     images: list[dict] | None = None,
+    context: list[dict] | None = None,
 ) -> ChatContext:
     registry = runtime.run_registry
 
@@ -187,7 +197,7 @@ async def prepare_chat(
         session_state.name = name_candidate[:50]
 
     messages = await _prepare_messages(
-        runtime, messages, user_message, last_activity=session_state.last_activity, images=images
+        runtime, messages, user_message, last_activity=session_state.last_activity, images=images, context=context
     )
 
     run = registry.create_run(session_state.session_id)