diff --git a/backend/apps/agents/agents.py b/backend/apps/agents/agents.py
index 50fd6773..9b6f3f45 100644
--- a/backend/apps/agents/agents.py
+++ b/backend/apps/agents/agents.py
@@ -34,6 +34,7 @@ from backend.apps.agents.session_store import (
load,
)
from backend.apps.agents import ws
+from backend.apps.agents.compose_system_prompt import compose_system_prompt
from claude_agent_sdk import ClaudeAgentOptions
SESSIONS: dict[str, Agent] = {}
@@ -51,7 +52,7 @@ async def p_send_browser_command(
action: str, browser_id: str, tab_id: str, params: dict,
) -> dict:
"""BrowserCommandFn implementation that routes through the browser FutureBridge."""
- request_id = uuid4().hex
+ request_id: str = uuid4().hex
if not ws.has_global_connections():
return {"error": "No dashboard connected. Open the dashboard to use browser tools."}
return await ws.BROWSER_BRIDGE.request(
@@ -133,20 +134,22 @@ class LaunchBody(BaseModel):
@agents.router.post("/launch")
async def launch(body: LaunchBody) -> dict:
- # TODO: build ClaudeAgentOptions from body once prompt/options builder exists
+ system_prompt = compose_system_prompt(
+ session_prompt=body.system_prompt or None,
+ )
agent: Agent = Agent(
model=body.model,
mode=body.mode,
- status="running",
+ status="stopped",
config=ClaudeAgentOptions(
- system_prompt=body.system_prompt,
+ system_prompt=system_prompt,
max_turns=body.max_turns,
),
)
agent.on_event = p_make_session_emitter(agent.session_id)
SESSIONS[agent.session_id] = agent
- await agent._emit(AgentStatusEvent(
- session_id=agent.session_id, status="running",
+ await agent.emit(AgentStatusEvent(
+ session_id=agent.session_id, status="stopped",
session=agent.snapshot(),
))
return {"session_id": agent.session_id, "session": agent.snapshot().model_dump(mode="json")}
@@ -163,7 +166,7 @@ async def update_session(session_id: str, body: UpdateBody) -> dict:
agent.name = body.name
if body.system_prompt is not None:
agent.config.system_prompt = body.system_prompt
- await agent._emit(AgentStatusEvent(
+ await agent.emit(AgentStatusEvent(
session_id=session_id, status=agent.status,
session=agent.snapshot(),
))
@@ -254,7 +257,7 @@ async def edit_message(session_id: str, body: EditMessageBody) -> dict:
fork: Agent = agent.branch(body.message_id)
SESSIONS[fork.session_id] = fork
- edited_msg = UserMessage(content=body.content, branch_id=fork.branch_id)
+ edited_msg: UserMessage = UserMessage(content=body.content, branch_id=fork.branch_id)
await fork.send_message(edited_msg)
return {"ok": True, "branch_id": fork.branch_id, "session_id": fork.session_id}
diff --git a/backend/apps/agents/compose_system_prompt.py b/backend/apps/agents/compose_system_prompt.py
new file mode 100644
index 00000000..32548bf3
--- /dev/null
+++ b/backend/apps/agents/compose_system_prompt.py
@@ -0,0 +1,30 @@
+"""Pure prompt-building helpers.
+
+All functions are stateless — they accept data as parameters and return
+strings. No imports from ``apps/`` or any external stores.
+"""
+
+from typing import Optional, List
+from typeguard import typechecked
+
+@typechecked
+def compose_system_prompt(
+ global_default: Optional[str] = None,
+ mode_prompt: Optional[str] = None,
+ session_prompt: Optional[str] = None,
+ connected_tools_ctx: Optional[str] = None,
+ browser_ctx: Optional[str] = None,
+) -> Optional[str]:
+ """Layer multiple prompt sources into one system prompt.
+
+ Order matters — earlier layers provide base context, later layers
+ override or augment.
+ """
+ parts: List[str] = [p for p in (
+ global_default,
+ mode_prompt,
+ session_prompt,
+ connected_tools_ctx,
+ browser_ctx,
+ ) if p]
+ return "\n\n".join(parts) if parts else None
\ No newline at end of file
diff --git a/backend/core/shared_structs/agent/Message/Message.py b/backend/core/shared_structs/agent/Message/Message.py
index 28b8fed1..d75e9cd1 100644
--- a/backend/core/shared_structs/agent/Message/Message.py
+++ b/backend/core/shared_structs/agent/Message/Message.py
@@ -12,6 +12,9 @@ from backend.core.shared_structs.agent.Message.agent_inputs import (
from backend.core.shared_structs.agent.Message.agent_outputs import (
ToolCallContent, ToolResultContent
)
+from backend.core.shared_structs.agent.Message.prompt_utils import (
+ resolve_context_paths, resolve_forced_tools, resolve_attached_skills,
+)
class Message(BaseModel):
id: str = Field(default_factory=lambda: uuid4().hex)
@@ -39,7 +42,15 @@ class UserMessage(Message):
@typechecked
def to_prompt(self) -> PromptMsgDict:
- blocks: List[PromptBlock] = [TextPromptBlock(type="text", text=self.content)]
+ parts: List[str] = [
+ resolve_forced_tools(self.forced_tools),
+ resolve_context_paths(self.context_paths),
+ resolve_attached_skills(self.attached_skills),
+ self.content,
+ ]
+ full_text: str = "\n\n".join(p for p in parts if p)
+
+ blocks: List[PromptBlock] = [TextPromptBlock(type="text", text=full_text)]
for data, media_type in zip[tuple[str, str]](self.images, self.image_media_types):
blocks.append(ImagePromptBlock(
type="image",
diff --git a/backend/core/shared_structs/agent/Message/agent_inputs.py b/backend/core/shared_structs/agent/Message/agent_inputs.py
index 5ffb22d2..b74c6f8b 100644
--- a/backend/core/shared_structs/agent/Message/agent_inputs.py
+++ b/backend/core/shared_structs/agent/Message/agent_inputs.py
@@ -30,4 +30,5 @@ class ContextPath(BaseModel):
class SkillMeta(BaseModel):
id: str
- name: str
\ No newline at end of file
+ name: str
+ content: str = ""
\ No newline at end of file
diff --git a/backend/core/shared_structs/agent/Message/prompt_utils.py b/backend/core/shared_structs/agent/Message/prompt_utils.py
new file mode 100644
index 00000000..d964806d
--- /dev/null
+++ b/backend/core/shared_structs/agent/Message/prompt_utils.py
@@ -0,0 +1,101 @@
+"""Pure prompt-building helpers.
+
+All functions are stateless — they accept data as parameters and return
+strings. No imports from ``apps/`` or any external stores.
+"""
+
+import os
+from typing import List
+
+from backend.core.shared_structs.agent.Message.agent_inputs import ContextPath, SkillMeta
+from typeguard import typechecked
+
+MAX_FILE_READ_BYTES = 512_000
+DIR_TREE_MAX_DEPTH = 4
+
+@typechecked
+def p_build_dir_tree(
+ root: str,
+ max_depth: int = DIR_TREE_MAX_DEPTH,
+ prefix: str = "",
+) -> List[str]:
+ lines: List[str] = []
+ try:
+ entries: List[str] = sorted(os.listdir(root))
+ except PermissionError:
+ return [f"{prefix}[permission denied]"]
+ dirs: List[str] = [
+ e for e in entries
+ if not e.startswith(".") and os.path.isdir(os.path.join(root, e))
+ ]
+ files: List[str] = [
+ e for e in entries
+ if not e.startswith(".") and os.path.isfile(os.path.join(root, e))
+ ]
+ for f in files:
+ lines.append(f"{prefix}{f}")
+ for d in dirs:
+ lines.append(f"{prefix}{d}/")
+ if max_depth > 1:
+ lines.extend(p_build_dir_tree(os.path.join(root, d), max_depth - 1, prefix + " "))
+ return lines
+
+
+@typechecked
+def resolve_context_paths(context_paths: List[ContextPath]) -> str:
+ """Read files / build directory trees for each context path."""
+ if not context_paths:
+ return ""
+ sections: List[str] = []
+ for cp in context_paths:
+ path: str = cp.path
+ cp_type: str = cp.type
+ if not path or not os.path.exists(path):
+ sections.append(f"[Context: {path} — not found]")
+ continue
+ if cp_type == "file" and os.path.isfile(path):
+ try:
+ with open(path, "r", errors="replace") as f:
+ content: str = f.read(MAX_FILE_READ_BYTES)
+ sections.append(
+ f'\n{content}\n'
+ )
+ except Exception as e:
+ sections.append(f"[Context: {path} — error reading: {e}]")
+ elif cp_type == "directory" and os.path.isdir(path):
+ tree = p_build_dir_tree(path, max_depth=DIR_TREE_MAX_DEPTH)
+ sections.append(
+ f'\n{chr(10).join(tree)}\n'
+ )
+ else:
+ sections.append(f"[Context: {path} — type mismatch]")
+ return "\n\n".join(sections)
+
+
+@typechecked
+def resolve_forced_tools(forced_tools: List[str]) -> str:
+ """Build a prompt block from a list of tool names."""
+ if not forced_tools:
+ return ""
+ lines: List[str] = [f"- {name}" for name in forced_tools]
+ return (
+ "\n"
+ "The user explicitly requested these tools be used. "
+ "Prioritize using them to address the user's request.\n"
+ + "\n".join(lines)
+ + "\n"
+ )
+
+
+@typechecked
+def resolve_attached_skills(attached_skills: List[SkillMeta]) -> str:
+ """Format attached skills into prompt sections."""
+ if not attached_skills:
+ return ""
+ sections: List[str] = []
+ for skill in attached_skills:
+ name: str = skill.name
+ content: str = skill.content
+ if content:
+ sections.append(f"[Using skill: {name}]\n\n{content}")
+ return "\n\n".join(sections)
\ No newline at end of file