apps/whatsapp_control_api/app/services/hermes_bridge.py
text
from __future__ import annotations
import asyncio
import logging
import os
import shlex
from collections.abc import Awaitable, Callable
from datetime import datetime, timezone
from pathlib import Path
from ..core.config import get_settings
logger = logging.getLogger(__name__)
ProgressCallback = Callable[[str, int, str], Awaitable[None]]
class HermesBridge:
@property
def settings(self):
return get_settings()
async def run_task(self, task_id: str, prompt: str, on_progress: ProgressCallback | None = None) -> tuple[int, str]:
task_dir = self.settings.task_logs_dir / task_id
task_dir.mkdir(parents=True, exist_ok=True)
prompt_path = task_dir / "prompt.md"
log_path = task_dir / "agent.log"
prompt_text = self._build_prompt(task_id, prompt)
prompt_path.write_text(prompt_text, encoding="utf-8")
command = shlex.split(self.settings.hermes_command)
if self.settings.hermes_args:
command.extend(shlex.split(self.settings.hermes_args))
if not command:
raise RuntimeError("HERMES_COMMAND is empty.")
exit_code, summary = await self._run_command(
task_id=task_id,
command=command,
prompt_text=prompt_text,
log_path=log_path,
on_progress=on_progress,
start_progress=5,
start_step="Starting Hermes Agent",
)
if exit_code == 0 and self._is_suspicious_summary(summary):
retry_prompt = self._build_retry_prompt(task_id, prompt)
(task_dir / "retry_prompt.md").write_text(retry_prompt, encoding="utf-8")
with log_path.open("a", encoding="utf-8") as log_file:
log_file.write("\n[control-layer] Suspiciously short Hermes output. Retrying with compact prompt.\n")
exit_code, retry_summary = await self._run_command(
task_id=task_id,
command=command,
prompt_text=retry_prompt,
log_path=log_path,
on_progress=on_progress,
start_progress=12,
start_step="Retrying Hermes final response",
)
if retry_summary.strip():
summary = retry_summary
return exit_code, summary
async def _run_command(
self,
*,
task_id: str,
command: list[str],
prompt_text: str,
log_path: Path,
on_progress: ProgressCallback | None,
start_progress: int,
start_step: str,
) -> tuple[int, str]:
process = await self._spawn_process(command, prompt_text)
assert process.stdout is not None
summary_lines: list[str] = []
current_progress = start_progress
current_step = start_step
started_at = asyncio.get_running_loop().time()
if on_progress:
await on_progress(task_id, current_progress, current_step)
with log_path.open("a", encoding="utf-8") as log_file:
while True:
try:
raw_line = await asyncio.wait_for(process.stdout.readline(), timeout=20)
except asyncio.TimeoutError:
if asyncio.get_running_loop().time() - started_at > self.settings.hermes_task_timeout_seconds:
process.kill()
await process.wait()
raise RuntimeError("Hermes task timed out.")
if current_progress < 90:
current_progress = min(current_progress + 3, 90)
current_step = "Hermes is working"
if on_progress:
await on_progress(task_id, current_progress, current_step)
continue
if not raw_line:
break
line = raw_line.decode("utf-8", errors="replace").rstrip()
log_file.write(line + "\n")
summary_lines.append(line)
if len(summary_lines) > 40:
summary_lines = summary_lines[-40:]
next_progress, next_step = self._advance(current_progress, current_step, line)
if next_progress != current_progress or next_step != current_step:
current_progress = next_progress
current_step = next_step
if on_progress:
await on_progress(task_id, current_progress, current_step)
try:
exit_code = await asyncio.wait_for(process.wait(), timeout=self.settings.hermes_task_timeout_seconds)
except asyncio.TimeoutError as exc:
process.kill()
await process.wait()
raise RuntimeError("Hermes task timed out.") from exc
summary = "\n".join(summary_lines).strip()
return exit_code, summary
async def _spawn_process(self, base_command: list[str], prompt_text: str) -> asyncio.subprocess.Process:
command = list(base_command)
executable = Path(command[0]).name.lower() if command else ""
command_line = " ".join(command).lower()
oneshot_requested = any(flag in command_line for flag in (" --oneshot", " -z "))
env = os.environ.copy()
env["HERMES_WRAPPER_DISABLE"] = "1"
if executable.startswith("hermes") and not oneshot_requested:
command.extend(["--oneshot", prompt_text])
return await asyncio.create_subprocess_exec(
*command,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
cwd=str(self.settings.hermes_workspace_root),
env=env,
)
process = await asyncio.create_subprocess_exec(
*command,
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
cwd=str(self.settings.hermes_workspace_root),
env=env,
)
if process.stdin is None:
raise RuntimeError("Failed to open Hermes stdin.")
process.stdin.write(prompt_text.encode("utf-8"))
await process.stdin.drain()
process.stdin.close()
return process
def _advance(self, progress: int, step: str, line: str) -> tuple[int, str]:
lower = line.lower()
phase_map = [
("analy", 15, "Analysing requirement"),
("plan", 25, "Planning implementation"),
("edit", 45, "Editing files"),
("writ", 45, "Writing files"),
("command", 55, "Running command"),
("test", 70, "Running tests"),
("debug", 78, "Debugging"),
("build", 82, "Building"),
("deploy", 90, "Deploying"),
("complete", 98, "Final validation"),
]
for keyword, mapped_progress, mapped_step in phase_map:
if keyword in lower and mapped_progress > progress:
return mapped_progress, mapped_step
if progress < 88:
return min(progress + 1, 88), step
return progress, step
def _is_suspicious_summary(self, summary: str) -> bool:
stripped = summary.strip()
if len(stripped) < 25:
return True
if stripped.lower() in {"ba", "baik", "ok", "oke"}:
return True
return False
def _build_retry_prompt(self, task_id: str, prompt: str) -> str:
return f"""Jawab atau kerjakan permintaan user sebagai Hermes Agent.
Task ID: {task_id}
Projects root: {self.settings.hermes_projects_root}
Aturan:
- Jika user bertanya, jawab langsung seperti AI assistant yang rapi.
- Jika user meminta pembuatan/perubahan project, kerjakan sampai selesai di /root/hermes-projects dan validasi hasilnya.
- Jika user meminta dokumen, PDF, laporan, proposal, spreadsheet, presentasi, ZIP, atau file lain, buat file nyata di /root/hermes-projects/AI Assistants Outputs/{task_id}.
- Untuk deliverable file, jangan hanya memberi teks untuk disalin. Cantumkan absolute path file akhir di summary.
- Untuk task project, jangan klaim selesai sebelum folder/file utama benar-benar dibuat dan perintah validasi yang relevan sudah dijalankan.
- Jika perlu menjalankan dev server, jalankan sebagai background process atau lakukan build/static validation agar proses tidak menggantung.
- Jika ada error, jelaskan error dan langkah yang sudah dilakukan.
- Jika task membutuhkan data terbaru, berita, harga, dokumentasi dependency, atau informasi faktual yang bisa berubah, gunakan internet/web search/tools yang tersedia dan cantumkan sumber URL di hasil akhir.
- Berikan jawaban akhir dalam bahasa Indonesia yang ringkas, jelas, dan aman dikirim lewat WhatsApp.
Permintaan user:
{prompt}
"""
def _build_prompt(self, task_id: str, prompt: str) -> str:
now = datetime.now(timezone.utc).isoformat()
return f"""# WhatsApp Control Layer Task
You are Hermes Agent. Work as a full autonomous agent, not a passive executor.
Task ID: {task_id}
Created at: {now}
Source channel: WhatsApp Cloud API
Response channel: WhatsApp
Projects root: {self.settings.hermes_projects_root}
Rules:
- Analyse the requirement before editing.
- Create a concise plan.
- Use project folders under /root/hermes-projects only.
- Run relevant commands, checks, tests, and validation.
- For project creation/editing tasks, do not claim completion until the project path exists, key files are written, and relevant validation/build/check commands have been run or clearly explained as unavailable.
- For document, PDF, report, proposal, spreadsheet, presentation, ZIP, or any other file deliverable, create real files under /root/hermes-projects/AI Assistants Outputs/{task_id}. Do not answer with copy-paste text only when the user asked for a file.
- For file deliverables, include every final absolute file path under /root/hermes-projects in the final summary so WhatsApp can expose download links.
- Do not start long-lived servers in the foreground. Use background execution or static validation/build checks.
- Do not expose secrets in logs or output.
- If the task creates or edits a project, explicitly state the final absolute project path under /root/hermes-projects.
- For current information, market/news/data research, package documentation, or facts that may change, use available internet/web search/tools and cite source URLs in the final summary.
- In the final summary, include key files changed, important commands run, validation results, and any access path the user should know.
- Return a concise final summary that is safe to send through WhatsApp.
- Write the final answer in clean Indonesian.
- Avoid terminal-style noise, meta commentary, and generic closers like "kalau kamu ingin saya bantu lagi".
- Prefer short paragraphs or flat bullets. Keep it compact and readable in WhatsApp.
User task:
{prompt}
"""