niklas/jecio
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: niklas:assistant/next-from-main
Branches Tags
niklas:main
niklas:assistant/next-from-main
niklas:assistant/learn-p1-aed8ee
..
compare: niklas:main
Branches Tags
niklas:assistant/next-from-main
niklas:assistant/learn-p1-aed8ee
niklas:main

No commits in common. "assistant/next-from-main" and "main" have entirely different histories.
assistant/ ... main

6 changed files with 1050 additions and 1102 deletions
Show stats Expand all files Collapse all files

1147
app.py
View file

File diff suppressed because it is too large Load diff

2
services/__init__.py
View file

@ -1,2 +0,0 @@
"""Service-layer modules for assistant backend."""

140
services/assistant_planning.py
View file

@ -1,140 +0,0 @@
from typing import Any, Awaitable, Callable, Dict, List, Optional
def _get(obj: Any, name: str, default: Any = None) -> Any:
return getattr(obj, name, default)
def build_assistant_plan_prompt(payload: Any, source_docs: List[Dict[str, Any]]) -> str:
constraints = _get(payload, "constraints", []) or []
constraint_lines = "\n".join(f"- {c}" for c in constraints) if constraints else "- None"
context_chunks = []
for d in source_docs:
src = d.get("_source", {}) or {}
context_chunks.append(
"\n".join(
[
f"concept_id: {src.get('concept_id', '')}",
f"source_pk: {src.get('source_pk', '')}",
f"source_table: {src.get('source_table', '')}",
f"release_name: {src.get('release_name', '')}",
f"text: {str(src.get('text') or src.get('description') or src.get('summary') or '')[:600]}",
]
)
)
context = "\n\n---\n\n".join(context_chunks) if context_chunks else "No retrieved context."
return (
"You are a cautious personal assistant planner. Produce an execution plan only; do not execute anything.\n"
"Return valid JSON ONLY with this exact shape:\n"
'{'
'"plan": ['
'{"step_id":"S1","title":"...","action_type":"research|draft|ask_user|prepare_data|review","requires_approval":true|false,"notes":"..."}'
"]"
"}\n"
f"Use at most {_get(payload, 'max_steps', 3)} steps.\n"
"Prefer safe read-only and draft actions first.\n\n"
f"Task type: {_get(payload, 'task_type', '')}\n"
f"Objective: {_get(payload, 'objective', '')}\n"
f"Constraints:\n{constraint_lines}\n\n"
"Retrieved context:\n"
f"{context}\n"
)
def fallback_plan(max_steps: int) -> List[Dict[str, Any]]:
return [
{
"step_id": "S1",
"title": "Gather relevant facts and constraints",
"action_type": "research",
"requires_approval": False,
"notes": "Review messages/concepts and identify required context.",
},
{
"step_id": "S2",
"title": "Draft a response or action proposal",
"action_type": "draft",
"requires_approval": False,
"notes": "Produce a concise draft aligned with objective and constraints.",
},
{
"step_id": "S3",
"title": "Request user confirmation before any external action",
"action_type": "ask_user",
"requires_approval": True,
"notes": "Do not send or execute changes until approved.",
},
][: max_steps]
def find_plan_step(plan: List[Any], step_id: str) -> Optional[Any]:
for s in plan:
if _get(s, "step_id", "") == step_id:
return s
return None
def is_high_risk_step(step: Any) -> bool:
text = f"{_get(step, 'title', '')} {_get(step, 'notes', '') or ''}".lower()
high_risk_terms = [
"send",
"submit",
"pay",
"payment",
"transfer",
"wire",
"sign",
"file",
"delete",
"close account",
"change account",
]
return any(t in text for t in high_risk_terms)
def enforce_step_policy(payload: Any, step: Any) -> Optional[str]:
if bool(_get(step, "requires_approval", False)) and not bool(_get(payload, "approved", False)):
return "Step requires approval but approved=false."
if is_high_risk_step(step):
if not bool(_get(payload, "approved", False)):
return "High-risk step requires approved=true."
token = str(_get(payload, "manual_confirm_token", "") or "")
if not token.strip():
return "High-risk step requires manual_confirm_token."
return None
async def execute_plan_step(
payload: Any,
step: Any,
generate_text: Callable[[str], Awaitable[str]],
) -> Dict[str, Any]:
action_type = str(_get(step, "action_type", ""))
title = str(_get(step, "title", ""))
notes = str(_get(step, "notes", "") or "")
if action_type == "draft":
prompt = (
"Draft concise text for this approved planning step.\n"
f"Task type: {_get(payload, 'task_type', '')}\n"
f"Objective: {_get(payload, 'objective', '')}\n"
f"Step: {title}\n"
f"Notes: {notes}\n"
"Output only final draft text."
)
try:
text = await generate_text(prompt)
if not text.strip():
text = f"Draft for step '{title}'."
except Exception:
text = f"Draft for step '{title}'."
return {"draft": text}
if action_type == "research":
return {"note": "Research step acknowledged. Use /search or /assistant/draft for grounded retrieval."}
if action_type == "prepare_data":
return {"note": "Prepare-data step acknowledged.", "checklist": ["Collect required inputs", "Normalize format", "Validate completeness"]}
if action_type == "review":
return {"note": "Review step requires human review before external action."}
if action_type == "ask_user":
return {"question": "Please confirm whether to proceed with the next high-impact action."}
return {"note": "Step recognized but no executor implemented."}

115
services/assistant_retrieval.py
View file

@ -1,115 +0,0 @@
import hashlib
import json
import re
from dataclasses import dataclass
from typing import Any, Awaitable, Callable, Dict, List, Optional, Set
DEFAULT_QUERY_STOPWORDS: Set[str] = {
"the", "and", "for", "with", "that", "this", "from", "have", "has", "had", "are", "was", "were", "will",
"would", "should", "could", "can", "you", "your", "about", "what", "when", "where", "which", "who", "whom",
"why", "how", "tomorrow", "today", "please", "any", "there", "need", "want", "know", "does", "did", "done",
}
@dataclass(frozen=True)
class AssistantRetrievalConfig:
min_token_overlap: int = 1
source_candidate_multiplier: int = 4
source_min_coverage: float = 0.6
query_stopwords: Set[str] = frozenset(DEFAULT_QUERY_STOPWORDS)
def query_tokens(text: str, stopwords: Set[str]) -> Set[str]:
return {
t for t in re.findall(r"[a-z0-9]{3,}", (text or "").lower())
if t not in stopwords
}
def source_text_for_match(src: Dict[str, Any]) -> str:
return " ".join(
[
str(src.get("text") or ""),
str(src.get("description") or ""),
str(src.get("summary") or ""),
str(src.get("display_name") or ""),
str(src.get("canonical_name") or ""),
]
)
def is_strong_source_match(query: str, src: Dict[str, Any], cfg: AssistantRetrievalConfig) -> bool:
q_tokens = query_tokens(query, cfg.query_stopwords)
if not q_tokens:
return True
s_tokens = query_tokens(source_text_for_match(src), cfg.query_stopwords)
overlap = len(q_tokens.intersection(s_tokens))
q_len = max(1, len(q_tokens))
coverage = overlap / q_len
min_overlap = cfg.min_token_overlap
if q_len >= 2:
min_overlap = max(min_overlap, 2)
return overlap >= min_overlap and coverage >= cfg.source_min_coverage
async def retrieve_sources_two_stage(
query: str,
release_name: Optional[str],
max_sources: int,
include_release_recent_fallback: bool,
cfg: AssistantRetrievalConfig,
es_search_hits: Callable[[str, int, Optional[str]], Awaitable[List[Dict[str, Any]]]],
es_recent_by_release: Callable[[str, int], Awaitable[List[Dict[str, Any]]]],
) -> List[Dict[str, Any]]:
candidate_size = max(max_sources, max_sources * max(2, cfg.source_candidate_multiplier))
seen_keys: set[str] = set()
candidates: List[Dict[str, Any]] = []
def add_hits(hs: List[Dict[str, Any]]) -> None:
for h in hs:
src = h.get("_source", {}) or {}
key = str(src.get("concept_id") or src.get("source_pk") or "")
if not key:
key = hashlib.sha256(
json.dumps(src, ensure_ascii=False, sort_keys=True).encode("utf-8")
).hexdigest()[:20]
if key in seen_keys:
continue
seen_keys.add(key)
candidates.append(h)
try:
add_hits(await es_search_hits(query, candidate_size, release_name))
except Exception as e:
print(f"[WARN] stage1 release search failed: {e}")
if len(candidates) < max_sources:
try:
add_hits(await es_search_hits(query, candidate_size, None))
except Exception as e:
print(f"[WARN] stage1 global search failed: {e}")
if len(candidates) < max_sources and include_release_recent_fallback and release_name:
try:
add_hits(await es_recent_by_release(release_name, candidate_size))
except Exception as e:
print(f"[WARN] stage1 release-recent fallback failed: {e}")
q_tokens = query_tokens(query, cfg.query_stopwords)
ranked: List[Dict[str, Any]] = []
for h in candidates:
src = h.get("_source", {}) or {}
s_tokens = query_tokens(source_text_for_match(src), cfg.query_stopwords)
overlap = len(q_tokens.intersection(s_tokens)) if q_tokens else 0
base_score = float(h.get("_score") or 0.0)
ranked.append({"hit": h, "overlap": overlap, "base_score": base_score})
ranked.sort(key=lambda x: (x["overlap"], x["base_score"]), reverse=True)
relevant = []
for x in ranked:
src = x["hit"].get("_source", {}) or {}
if is_strong_source_match(query, src, cfg):
relevant.append(x)
if relevant:
return [x["hit"] for x in relevant[:max_sources]]
return []

193
services/imap_ingest.py
View file

@ -1,193 +0,0 @@
import email
import hashlib
import imaplib
import os
from datetime import timezone
from email import policy
from email.utils import parseaddr, parsedate_to_datetime
from typing import Any, Dict, List, Optional
def clean_header_id(v: Optional[str]) -> str:
if not v:
return ""
return v.strip().strip("<>").strip()
def normalize_thread_id(msg_id: str, refs: str, in_reply_to: str, subject: str, sender: str) -> str:
refs_clean = clean_header_id(refs.split()[-1] if refs else "")
in_reply_clean = clean_header_id(in_reply_to)
if refs_clean:
return f"thread:{refs_clean}"
if in_reply_clean:
return f"thread:{in_reply_clean}"
seed = f"{subject.strip().lower()}|{sender.strip().lower()}"
if not seed.strip("|"):
seed = msg_id
return "thread:" + hashlib.sha256(seed.encode("utf-8")).hexdigest()[:24]
def extract_body_text(msg: email.message.Message) -> str:
try:
if msg.is_multipart():
for part in msg.walk():
ctype = (part.get_content_type() or "").lower()
disp = (part.get("Content-Disposition") or "").lower()
if ctype == "text/plain" and "attachment" not in disp:
payload_obj = part.get_content()
if isinstance(payload_obj, str):
return payload_obj.strip()
if isinstance(payload_obj, bytes):
return payload_obj.decode(part.get_content_charset() or "utf-8", errors="replace").strip()
for part in msg.walk():
ctype = (part.get_content_type() or "").lower()
if ctype == "text/html":
html_obj = part.get_content()
if isinstance(html_obj, bytes):
html_obj = html_obj.decode(part.get_content_charset() or "utf-8", errors="replace")
if isinstance(html_obj, str):
return html_obj.strip()
return ""
payload_obj = msg.get_content()
if isinstance(payload_obj, str):
return payload_obj.strip()
if isinstance(payload_obj, bytes):
return payload_obj.decode(msg.get_content_charset() or "utf-8", errors="replace").strip()
return ""
except Exception:
return ""
def fetch_imap_messages_blocking(
payload: Any,
effective_search_criteria: str,
since_uid: Optional[int],
) -> List[Dict[str, Any]]:
password = getattr(payload, "password", None) or os.getenv("IMAP_PASSWORD", "")
if not password:
raise ValueError("IMAP password missing: provide payload.password or set IMAP_PASSWORD")
host = str(getattr(payload, "host"))
port = int(getattr(payload, "port"))
use_ssl = bool(getattr(payload, "use_ssl"))
username = str(getattr(payload, "username"))
mailbox = str(getattr(payload, "mailbox"))
max_messages = int(getattr(payload, "max_messages"))
channel = str(getattr(payload, "channel"))
if use_ssl:
client = imaplib.IMAP4_SSL(host, port)
else:
client = imaplib.IMAP4(host, port)
try:
status, _ = client.login(username, password)
if status != "OK":
raise RuntimeError("IMAP login failed")
status, _ = client.select(mailbox, readonly=True)
if status != "OK":
raise RuntimeError(f"IMAP select mailbox failed: {mailbox}")
if since_uid is not None:
status, search_data = client.uid("search", None, "UID", f"{int(since_uid) + 1}:*")
else:
status, search_data = client.uid("search", None, effective_search_criteria)
if status != "OK":
raise RuntimeError(f"IMAP search failed: {effective_search_criteria}")
uid_bytes = search_data[0] if search_data else b""
uid_list = [u for u in uid_bytes.decode("utf-8", errors="replace").split() if u]
if since_uid is not None:
filtered: List[str] = []
for u in uid_list:
try:
if int(u) > int(since_uid):
filtered.append(u)
except Exception:
continue
uid_list = filtered
if not uid_list:
return []
is_uid_window = since_uid is not None
if is_uid_window:
selected_uids = uid_list[:max_messages]
else:
selected_uids = uid_list[-max_messages:]
out: List[Dict[str, Any]] = []
for uid in selected_uids:
status, msg_data = client.uid("fetch", uid, "(RFC822)")
if status != "OK" or not msg_data:
continue
raw_bytes = None
for part in msg_data:
if isinstance(part, tuple) and len(part) >= 2 and isinstance(part[1], (bytes, bytearray)):
raw_bytes = bytes(part[1])
break
if not raw_bytes:
continue
msg = email.message_from_bytes(raw_bytes, policy=policy.default)
subject = str(msg.get("Subject") or "").strip()
from_raw = str(msg.get("From") or "").strip()
to_raw = str(msg.get("To") or "").strip()
date_raw = str(msg.get("Date") or "").strip()
msg_id_raw = str(msg.get("Message-Id") or msg.get("Message-ID") or "").strip()
refs_raw = str(msg.get("References") or "").strip()
in_reply_raw = str(msg.get("In-Reply-To") or "").strip()
sender_email = parseaddr(from_raw)[1] or from_raw or "unknown"
msg_id_clean = clean_header_id(msg_id_raw)
if not msg_id_clean:
seed = f"{uid}|{subject}|{sender_email}|{date_raw}"
msg_id_clean = "imap-" + hashlib.sha256(seed.encode("utf-8")).hexdigest()[:24]
thread_id = normalize_thread_id(
msg_id=msg_id_clean,
refs=refs_raw,
in_reply_to=in_reply_raw,
subject=subject,
sender=sender_email,
)
sent_at_iso = None
if date_raw:
try:
dt = parsedate_to_datetime(date_raw)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
sent_at_iso = dt.astimezone(timezone.utc).isoformat()
except Exception:
sent_at_iso = None
body = extract_body_text(msg)
if not body:
body = f"(no body) {subject}".strip()
metadata = {
"subject": subject,
"from": from_raw,
"to": to_raw,
"date": date_raw,
"imap_uid": uid,
"mailbox": mailbox,
"host": host,
"username": username,
}
out.append(
{
"thread_id": thread_id,
"message_id": msg_id_clean,
"sender": sender_email,
"channel": channel,
"sent_at": sent_at_iso,
"body": body,
"metadata": metadata,
}
)
return out
finally:
try:
client.logout()
except Exception:
pass

555
services/remote_ops.py
View file

@ -1,555 +0,0 @@
import asyncio
import base64
import json
import os
import shlex
import tempfile
import uuid
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from fastapi import HTTPException
@dataclass(frozen=True)
class RemoteOpsConfig:
ssh_host: str
remote_dir: str
ssh_bin: str
ssh_opts: str
scp_bin: str
scp_opts: str
timeout_sec: int
projector_remote_script: str
ingest_message_remote_script: str
ingest_messages_batch_remote_script: str
assistant_feedback_remote_script: str
assistant_feedback_query_remote_script: str
assistant_metrics_query_remote_script: str
assistant_action_remote_script: str
assistant_actions_query_remote_script: str
assistant_proposals_remote_script: str
assistant_proposals_query_remote_script: str
runs_remote_script: str
run_events_remote_script: str
imap_checkpoint_remote_script: str
create_messages_release_remote_script: str
def _tail(text: str, max_chars: int = 8000) -> str:
if len(text) <= max_chars:
return text
return text[-max_chars:]
def _b64(s: str) -> str:
return base64.b64encode(s.encode("utf-8")).decode("ascii")
def _extract_json_array_from_text(text: str) -> List[Dict[str, Any]]:
start = text.find("[")
end = text.rfind("]")
if start == -1 or end == -1 or end < start:
raise ValueError("No JSON array found in output")
candidate = text[start : end + 1]
obj = json.loads(candidate)
if not isinstance(obj, list):
raise ValueError("Parsed value is not a JSON array")
out: List[Dict[str, Any]] = []
for item in obj:
if isinstance(item, dict):
out.append(item)
return out
def _extract_json_object_from_text(text: str) -> Dict[str, Any]:
start = text.find("{")
end = text.rfind("}")
if start == -1 or end == -1 or end < start:
raise ValueError("No JSON object found in output")
candidate = text[start : end + 1]
obj = json.loads(candidate)
if not isinstance(obj, dict):
raise ValueError("Parsed value is not a JSON object")
return obj
class RemoteOps:
def __init__(self, cfg: RemoteOpsConfig):
self.cfg = cfg
def _ssh_args(self, command: str) -> List[str]:
return [self.cfg.ssh_bin, *shlex.split(self.cfg.ssh_opts), self.cfg.ssh_host, command]
async def _run_ssh(self, parts: List[str], timeout_error: str) -> Dict[str, Any]:
command = f"cd {shlex.quote(self.cfg.remote_dir)} && {' '.join(shlex.quote(p) for p in parts)}"
proc = await asyncio.create_subprocess_exec(
*self._ssh_args(command),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.cfg.timeout_sec)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise HTTPException(status_code=504, detail=timeout_error)
out = stdout.decode("utf-8", errors="replace")
err = stderr.decode("utf-8", errors="replace")
return {"code": proc.returncode, "out": out, "err": err}
def _error_detail(self, code: int, out: str, err: str) -> Dict[str, Any]:
return {
"host": self.cfg.ssh_host,
"remote_dir": self.cfg.remote_dir,
"exit_code": code,
"stdout_tail": _tail(out),
"stderr_tail": _tail(err),
}
async def run_remote_query_imap_checkpoint(
self,
host: str,
mailbox: str,
username: str,
table: str,
) -> Optional[int]:
res = await self._run_ssh(
[self.cfg.imap_checkpoint_remote_script, host, mailbox, username, table],
"IMAP checkpoint query timed out",
)
if res["code"] != 0:
raise HTTPException(status_code=502, detail=self._error_detail(res["code"], res["out"], res["err"]))
try:
obj = _extract_json_object_from_text(res["out"])
val = obj.get("max_uid")
if val is None:
return None
return int(val)
except Exception as e:
raise HTTPException(
status_code=502,
detail={
"message": f"Unable to parse IMAP checkpoint output: {e}",
"stdout_tail": _tail(res["out"]),
"stderr_tail": _tail(res["err"]),
},
)
async def run_remote_create_messages_release(self, release_name: str) -> Dict[str, Any]:
res = await self._run_ssh(
[self.cfg.create_messages_release_remote_script, release_name],
"Create messages release timed out",
)
result = {
**self._error_detail(res["code"], res["out"], res["err"]),
"release_name": release_name,
}
if res["code"] != 0:
raise HTTPException(status_code=502, detail=result)
return result
async def run_remote_projector(self, payload: Any) -> Dict[str, Any]:
parts = [
self.cfg.projector_remote_script,
"--release-name", str(getattr(payload, "release_name", "")),
"--targets", str(getattr(payload, "targets", "both")),
]
if getattr(payload, "concept_table", None):
parts.extend(["--concept-table", str(getattr(payload, "concept_table"))])
if bool(getattr(payload, "dry_run", False)):
parts.append("--dry-run")
res = await self._run_ssh(parts, "Projector execution timed out")
result = {
**self._error_detail(res["code"], res["out"], res["err"]),
"spark_read_done": "[STEP] spark_read_done" in res["out"],
"projection_done": "[STEP] projection_done" in res["out"],
}
if res["code"] != 0:
raise HTTPException(status_code=502, detail=result)
return result
async def run_remote_ingest_message(self, payload: Any) -> Dict[str, Any]:
parts = [
self.cfg.ingest_message_remote_script,
str(getattr(payload, "table")),
str(getattr(payload, "thread_id")),
str(getattr(payload, "message_id")),
str(getattr(payload, "sender")),
str(getattr(payload, "channel")),
str(getattr(payload, "sent_at") or ""),
_b64(str(getattr(payload, "body") or "")),
_b64(json.dumps(getattr(payload, "metadata", {}) or {}, ensure_ascii=False)),
]
res = await self._run_ssh(parts, "Message ingest execution timed out")
result = self._error_detail(res["code"], res["out"], res["err"])
if res["code"] != 0:
raise HTTPException(status_code=502, detail=result)
return result
async def run_remote_ingest_messages_batch(self, payload: Any) -> Dict[str, Any]:
rows = []
for m in list(getattr(payload, "messages", []) or []):
rows.append(
{
"thread_id": getattr(m, "thread_id"),
"message_id": getattr(m, "message_id"),
"sender": getattr(m, "sender"),
"channel": getattr(m, "channel"),
"sent_at": getattr(m, "sent_at"),
"body": getattr(m, "body"),
"metadata": getattr(m, "metadata"),
}
)
if not rows:
return {
"host": self.cfg.ssh_host,
"remote_dir": self.cfg.remote_dir,
"exit_code": 0,
"rows": 0,
"stdout_tail": "[INFO] No rows to ingest",
"stderr_tail": "",
}
local_tmp = tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False, encoding="utf-8")
remote_tmp = f"{self.cfg.remote_dir}/.ingest_messages_{uuid.uuid4().hex}.json"
try:
json.dump(rows, local_tmp, ensure_ascii=False)
local_tmp.flush()
local_tmp.close()
scp_target = f"{self.cfg.ssh_host}:{remote_tmp}"
scp_args = [self.cfg.scp_bin, *shlex.split(self.cfg.scp_opts), local_tmp.name, scp_target]
scp_proc = await asyncio.create_subprocess_exec(
*scp_args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
scp_stdout, scp_stderr = await asyncio.wait_for(scp_proc.communicate(), timeout=self.cfg.timeout_sec)
except asyncio.TimeoutError:
scp_proc.kill()
await scp_proc.wait()
raise HTTPException(status_code=504, detail="Batch payload upload timed out")
if scp_proc.returncode != 0:
raise HTTPException(
status_code=502,
detail={
"host": self.cfg.ssh_host,
"remote_dir": self.cfg.remote_dir,
"exit_code": scp_proc.returncode,
"stdout_tail": _tail(scp_stdout.decode("utf-8", errors="replace")),
"stderr_tail": _tail(scp_stderr.decode("utf-8", errors="replace")),
},
)
payload_arg = f"@{remote_tmp}"
parts = [
self.cfg.ingest_messages_batch_remote_script,
str(getattr(payload, "table")),
str(getattr(payload, "dedupe_mode")),
payload_arg,
]
batch_cmd = " ".join(shlex.quote(p) for p in parts)
command = (
f"cd {shlex.quote(self.cfg.remote_dir)} && "
f"({batch_cmd}); rc=$?; rm -f {shlex.quote(remote_tmp)}; exit $rc"
)
proc = await asyncio.create_subprocess_exec(
*self._ssh_args(command),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
try:
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.cfg.timeout_sec)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise HTTPException(status_code=504, detail="Batch message ingest execution timed out")
finally:
try:
os.unlink(local_tmp.name)
except Exception:
pass
out = stdout.decode("utf-8", errors="replace")
err = stderr.decode("utf-8", errors="replace")
result = {
"host": self.cfg.ssh_host,
"remote_dir": self.cfg.remote_dir,
"exit_code": proc.returncode,
"rows": len(rows),
"stdout_tail": _tail(out),
"stderr_tail": _tail(err),
}
if proc.returncode != 0:
raise HTTPException(status_code=502, detail=result)
return result
async def run_remote_assistant_feedback(self, feedback_id: str, payload: Any, created_at_utc: str) -> Dict[str, Any]:
confidence = getattr(payload, "confidence", None)
conf = confidence if confidence is not None else 0.0
sources = [s.model_dump() for s in list(getattr(payload, "sources", []) or [])]
parts = [
self.cfg.assistant_feedback_remote_script,
feedback_id,
created_at_utc,
str(getattr(payload, "outcome")),
str(getattr(payload, "task_type")),
str(getattr(payload, "release_name") or ""),
f"{conf}",
"true" if bool(getattr(payload, "needs_review", False)) else "false",
_b64(str(getattr(payload, "goal") or "")),
_b64(str(getattr(payload, "draft") or "")),
_b64(str(getattr(payload, "final_text") or "")),
_b64(json.dumps(sources, ensure_ascii=False)),
_b64(str(getattr(payload, "notes") or "")),
]
res = await self._run_ssh(parts, "Assistant feedback execution timed out")
result = self._error_detail(res["code"], res["out"], res["err"])
if res["code"] != 0:
raise HTTPException(status_code=502, detail=result)
return result
async def run_remote_query_assistant_feedback(
self, outcome: Optional[str], task_type: Optional[str], release_name: Optional[str], limit: int
) -> Dict[str, Any]:
parts = [
self.cfg.assistant_feedback_query_remote_script,
outcome or "",
task_type or "",
release_name or "",
str(limit),
]
res = await self._run_ssh(parts, "Assistant feedback query timed out")
if res["code"] != 0:
raise HTTPException(status_code=502, detail=self._error_detail(res["code"], res["out"], res["err"]))
try:
rows = _extract_json_array_from_text(res["out"])
except Exception as e:
raise HTTPException(
status_code=502,
detail={
"message": f"Unable to parse feedback query output: {e}",
"stdout_tail": _tail(res["out"]),
"stderr_tail": _tail(res["err"]),
},
)
return {"host": self.cfg.ssh_host, "remote_dir": self.cfg.remote_dir, "rows": rows}
async def run_remote_query_assistant_metrics(
self, task_type: Optional[str], release_name: Optional[str], outcome: Optional[str], group_by: str, limit: int
) -> Dict[str, Any]:
parts = [
self.cfg.assistant_metrics_query_remote_script,
task_type or "",
release_name or "",
outcome or "",
group_by,
str(limit),
]
res = await self._run_ssh(parts, "Assistant metrics query timed out")
if res["code"] != 0:
raise HTTPException(status_code=502, detail=self._error_detail(res["code"], res["out"], res["err"]))
try:
rows = _extract_json_array_from_text(res["out"])
except Exception as e:
raise HTTPException(
status_code=502,
detail={
"message": f"Unable to parse metrics query output: {e}",
"stdout_tail": _tail(res["out"]),
"stderr_tail": _tail(res["err"]),
},
)
return {"host": self.cfg.ssh_host, "remote_dir": self.cfg.remote_dir, "rows": rows}
async def run_remote_assistant_action(
self, action_id: str, payload: Any, step: Any, status: str, output_json: Dict[str, Any], error_text: Optional[str], created_at_utc: str
) -> Dict[str, Any]:
parts = [
self.cfg.assistant_action_remote_script,
action_id,
created_at_utc,
str(getattr(payload, "task_type")),
str(getattr(payload, "release_name") or ""),
_b64(str(getattr(payload, "objective") or "")),
str(getattr(step, "step_id")),
_b64(str(getattr(step, "title") or "")),
str(getattr(step, "action_type")),
"true" if bool(getattr(step, "requires_approval", False)) else "false",
"true" if bool(getattr(payload, "approved", False)) else "false",
status,
_b64(json.dumps(output_json, ensure_ascii=False)),
_b64(error_text or ""),
]
res = await self._run_ssh(parts, "Assistant action logging timed out")
result = self._error_detail(res["code"], res["out"], res["err"])
if res["code"] != 0:
raise HTTPException(status_code=502, detail=result)
return result
async def run_remote_query_assistant_actions(
self,
status: Optional[str],
task_type: Optional[str],
release_name: Optional[str],
step_id: Optional[str],
action_type: Optional[str],
limit: int,
) -> Dict[str, Any]:
parts = [
self.cfg.assistant_actions_query_remote_script,
status or "",
task_type or "",
release_name or "",
step_id or "",
action_type or "",
str(limit),
]
res = await self._run_ssh(parts, "Assistant actions query timed out")
if res["code"] != 0:
raise HTTPException(status_code=502, detail=self._error_detail(res["code"], res["out"], res["err"]))
try:
rows = _extract_json_array_from_text(res["out"])
except Exception as e:
raise HTTPException(
status_code=502,
detail={
"message": f"Unable to parse actions query output: {e}",
"stdout_tail": _tail(res["out"]),
"stderr_tail": _tail(res["err"]),
},
)
return {"host": self.cfg.ssh_host, "remote_dir": self.cfg.remote_dir, "rows": rows}
async def run_remote_record_assistant_proposals(
self,
proposal_set_id: str,
created_at_utc: str,
objective: str,
release_name: Optional[str],
summary: str,
signals: Dict[str, Any],
proposals: List[Dict[str, Any]],
) -> Dict[str, Any]:
parts = [
self.cfg.assistant_proposals_remote_script,
proposal_set_id,
created_at_utc,
_b64(objective or ""),
release_name or "",
_b64(summary or ""),
_b64(json.dumps(signals or {}, ensure_ascii=False)),
_b64(json.dumps(proposals or [], ensure_ascii=False)),
]
res = await self._run_ssh(parts, "Assistant proposals logging timed out")
result = self._error_detail(res["code"], res["out"], res["err"])
if res["code"] != 0:
raise HTTPException(status_code=502, detail=result)
return result
async def run_remote_query_assistant_proposals(
self,
release_name: Optional[str],
proposal_set_id: Optional[str],
limit: int,
) -> Dict[str, Any]:
parts = [
self.cfg.assistant_proposals_query_remote_script,
release_name or "",
proposal_set_id or "",
str(limit),
]
res = await self._run_ssh(parts, "Assistant proposals query timed out")
if res["code"] != 0:
raise HTTPException(status_code=502, detail=self._error_detail(res["code"], res["out"], res["err"]))
try:
rows = _extract_json_array_from_text(res["out"])
except Exception as e:
raise HTTPException(
status_code=502,
detail={
"message": f"Unable to parse proposals query output: {e}",
"stdout_tail": _tail(res["out"]),
"stderr_tail": _tail(res["err"]),
},
)
return {"host": self.cfg.ssh_host, "remote_dir": self.cfg.remote_dir, "rows": rows}
async def run_remote_record_run(
self,
run_id: str,
run_type: str,
status: str,
started_at_utc: str,
finished_at_utc: str,
actor: str,
input_json: Dict[str, Any],
output_json: Optional[Dict[str, Any]],
error_text: Optional[str],
) -> None:
parts = [
self.cfg.runs_remote_script,
run_id,
run_type,
status,
started_at_utc,
finished_at_utc,
actor,
_b64(json.dumps(input_json, ensure_ascii=False)),
_b64(json.dumps(output_json, ensure_ascii=False) if output_json is not None else ""),
_b64(error_text or ""),
]
command = f"cd {shlex.quote(self.cfg.remote_dir)} && {' '.join(shlex.quote(p) for p in parts)}"
proc = await asyncio.create_subprocess_exec(
*self._ssh_args(command),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.cfg.timeout_sec)
if proc.returncode != 0:
raise HTTPException(
status_code=502,
detail={
"message": "Failed to record run in Iceberg",
"host": self.cfg.ssh_host,
"exit_code": proc.returncode,
"stdout_tail": _tail(stdout.decode("utf-8", errors="replace")),
"stderr_tail": _tail(stderr.decode("utf-8", errors="replace")),
},
)
async def run_remote_record_event(self, run_id: str, event_type: str, detail_json: Dict[str, Any], created_at_utc: str) -> None:
parts = [
self.cfg.run_events_remote_script,
run_id,
event_type,
created_at_utc,
_b64(json.dumps(detail_json, ensure_ascii=False)),
]
command = f"cd {shlex.quote(self.cfg.remote_dir)} && {' '.join(shlex.quote(p) for p in parts)}"
proc = await asyncio.create_subprocess_exec(
*self._ssh_args(command),
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=self.cfg.timeout_sec)
if proc.returncode != 0:
raise HTTPException(
status_code=502,
detail={
"message": "Failed to record run event in Iceberg",
"host": self.cfg.ssh_host,
"exit_code": proc.returncode,
"stdout_tail": _tail(stdout.decode("utf-8", errors="replace")),
"stderr_tail": _tail(stderr.decode("utf-8", errors="replace")),
},
)
async def record_event_best_effort(self, run_id: str, event_type: str, detail_json: Dict[str, Any], created_at_utc: str) -> None:
try:
await self.run_remote_record_event(run_id, event_type, detail_json, created_at_utc)
except Exception as e:
print(f"[WARN] run event logging failed: run_id={run_id} event={event_type} error={e}")