diff --git a/.github/dependabot.yml b/.github/dependabot.yml index 3854c8f9302f..bd8f47b3d7f8 100644 --- a/.github/dependabot.yml +++ b/.github/dependabot.yml @@ -42,3 +42,21 @@ updates: update-types: - "minor" - "patch" + + # gitsubmodule is the other exception: this fork carries the + # `tinker-atropos` submodule, which is a vendored pin to a specific + # upstream commit. Dependabot opens a PR when that upstream advances + # so we can update the pin deliberately, on review — same model as + # github-actions above. Monthly cadence is intentional: submodule + # bumps are heavier-weight than action SHA bumps and don't need a + # weekly rhythm. + - package-ecosystem: "gitsubmodule" + directory: "/" + schedule: + interval: "monthly" + labels: + - "dependencies" + - "submodule" + commit-message: + prefix: "chore(submodule)" + include: "scope" diff --git a/AGENTS.md b/AGENTS.md index b77a1d269990..0b32cf223f67 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -741,6 +741,69 @@ in config.yaml (or `HERMES_BACKGROUND_NOTIFICATIONS` env var): - `error` — only the final message when exit code != 0 - `off` — no watcher messages at all +### Optional extras and submodules — security boundaries + +A subset of opt-in extras and the `tinker-atropos` git submodule pull research +code from upstream repositories that **lack a LICENSE file** or are otherwise +unsuitable for production. These are deliberately fenced behind opt-in +extras / uninitialized submodules, and the boundary is load-bearing: if it is +crossed by accident (e.g. a deploy script that runs `pip install -e ".[all,yc-bench]"`), +unlicensed code lands in product runtime. The 2026-05-05 dependency audit +(`CHG-0001`) flagged this as the cheapest mitigation for two latent CRITICAL +findings. + +**Research-only — must NOT appear in any orchestrator default-install path, +deployment image, or CI workflow that runs in production:** + +| Item | Source | License | Notes | +|---|---|---|---| +| `[rl]` extra | `NousResearch/atropos` (MIT) + `thinking-machines-lab/tinker` (Apache-2.0) | OK individually | RL training stack — research-only, heavy GPU deps (`wandb`, `fastapi`). Not for product runtime. | +| `[yc-bench]` extra | `collinear-ai/yc-bench` @ `bfb0c88` | **NO LICENSE** (CRIT-2) | No grant of use. Pulling this into a deploy is an unlicensed-code incident. | +| `tinker-atropos` submodule | `nousresearch/tinker-atropos` | **NO LICENSE** (CRIT-1) | Registered in `.gitmodules`, intentionally **uninitialized**. Do NOT add `git submodule update --init` to any deploy/CI step. | + +**The operating rule:** + +1. The default install — `uv pip install .` (or `pip install -e .`, or + `uv pip install '.[all]'`) — does NOT pull `[rl]` or `[yc-bench]`. The + `[all]` meta-extra deliberately omits both. Keep it that way. +2. The `tinker-atropos` submodule must NOT be initialized in any + deployment-time path. `git clone` without `--recurse-submodules`, and + leave it that way. +3. Opting in is a deliberate research-mode choice. To enable the RL stack + locally for research, run `uv pip install '.[rl]'` (and `'.[yc-bench]'` + separately on Python ≥ 3.12). Doing so is an explicit acceptance of the + unlicensed-code risk for `yc-bench` / `tinker-atropos` and means the + environment is for personal research only — never a shared image, never + a deploy target. +4. Orchestrators that consume `hermes-agent-ucpm` (e.g. `paperclip-UCPM-orchestrator`) + MUST NOT add `[rl]` or `[yc-bench]` to their default-install manifest, and + MUST NOT initialize the `tinker-atropos` submodule in their build steps. + +**Maintainer check — verify defaults are clean:** + +```bash +# 1. Confirm [rl] and [yc-bench] are in [project.optional-dependencies] only, +# NOT in [project.dependencies] or [all]: +grep -nE '^(rl|yc-bench)\s*=' pyproject.toml # must be under [project.optional-dependencies] +grep -nE 'hermes-agent\[(rl|yc-bench)\]' pyproject.toml # must return nothing (i.e. not in [all]) + +# 2. Confirm tinker-atropos is registered but uninitialized in fresh clones: +git config --file .gitmodules --get submodule.tinker-atropos.url # registered → URL prints +ls tinker-atropos 2>/dev/null && echo "WARNING: submodule populated" || echo "OK: not initialized" + +# 3. Confirm the default install resolves without research deps: +uv pip install --dry-run . 2>&1 | grep -E '(yc-bench|atroposlib|tinker[^-])' \ + && echo "FAIL: research dep in default tree" || echo "OK: default tree clean" +``` + +If any of those checks flag a regression, the security boundary has been +breached — fix `pyproject.toml` / `.gitmodules` before merging. + +**Aside — `[messaging]`, `[matrix]`, `[homeassistant]`, `[sms]`:** these extras +pull `aiohttp@3.13.3`, which has 10 known CVEs (3 HIGH per audit CRIT-5). +Orchestrators layering these in should pin a `aiohttp>=3.14` floor in their +own manifest until the upstream pin is bumped. + --- ## Profiles: Multi-Instance Support diff --git a/CLAUDE_CODE.md b/CLAUDE_CODE.md new file mode 100644 index 000000000000..8e27cbfd2731 --- /dev/null +++ b/CLAUDE_CODE.md @@ -0,0 +1,11 @@ +# Claude Code Worktree + +This directory is a Claude Code worktree for `hermes-agent-ucpm`. + +- **Branch**: `chore/dependabot-gitsubmodule-CHG-0001` +- **Scope**: CHG-0001 — append `gitsubmodule` ecosystem block to `.github/dependabot.yml`. +- **Upstream policy preserved**: the existing `github-actions` block (and the comment block explaining why pip is intentionally excluded under uv.lock pinning) is untouched. Only the new `gitsubmodule` ecosystem is added. +- **Why a fork-only addition**: this fork carries the `tinker-atropos` submodule that upstream's policy doesn't cover. +- **Orchestrator issue**: https://github.com/WanderingStardust79/paperclip-UCPM-orchestrator/issues/2 + +Do not merge from this worktree directly — review and merge via PR on GitHub. diff --git a/hermes_agent/__init__.py b/hermes_agent/__init__.py new file mode 100644 index 000000000000..655b8f81dcef --- /dev/null +++ b/hermes_agent/__init__.py @@ -0,0 +1,10 @@ +"""UCPM-specific code for the hermes agent runtime. + +This package is namespaced separately from the upstream `hermes`/`hermes_cli` +modules so it can be carried through upstream rebases without conflicts. +Everything UCPM-specific (per-property loops, SOP-driven procedures, the +agent personas defined in `paperclip-UCPM/companies/ucpm-default/SOP.md`) +lives here. +""" + +__version__ = "0.1.0" diff --git a/hermes_agent/loops/__init__.py b/hermes_agent/loops/__init__.py new file mode 100644 index 000000000000..e59eaceefcce --- /dev/null +++ b/hermes_agent/loops/__init__.py @@ -0,0 +1,6 @@ +"""Per-property agent loops. + +Each loop is a small, file-driven runtime that exercises a subset of the +UCPM SOP procedures end-to-end without depending on Postgres, IMAP/SMTP, +or external services. Real I/O is wired in later. +""" diff --git a/hermes_agent/loops/audit.py b/hermes_agent/loops/audit.py new file mode 100644 index 000000000000..bfd77f447143 --- /dev/null +++ b/hermes_agent/loops/audit.py @@ -0,0 +1,95 @@ +"""Audit log writer for the file-based test property loop. + +One JSONL row per procedure step, schema mirrors `ucpm.audit_log` from the +canonical SOP global invariants: + ts, property_id, procedure_id, step, persona, inputs_hash, + action, output_ref, escalated_bool, operator_review_state. + +The on-disk shape is stable so a future BigQuery loader can ingest these +files unchanged. +""" + +from __future__ import annotations + +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional + +from .schemas import AuditRow + + +def compute_inputs_hash(payload: Any) -> str: + """Deterministic content hash of an arbitrary JSON-serializable payload. + + Used in `audit_row.inputs_hash` so replays are detectable and the + BigQuery `audit_log` table can dedupe by content. + """ + serialized = json.dumps(payload, sort_keys=True, default=str, separators=(",", ":")) + return hashlib.sha256(serialized.encode("utf-8")).hexdigest() + + +class AuditWriter: + """Append-only JSONL writer scoped to a single inbound message id. + + Each call to `write()` adds one line to `/.jsonl`. + The writer never reads or rewrites prior lines — appending only. + """ + + def __init__(self, audit_dir: Path, msg_id: str, property_id: str): + self.audit_dir = audit_dir + self.msg_id = msg_id + self.property_id = property_id + audit_dir.mkdir(parents=True, exist_ok=True) + self._path = audit_dir / f"{msg_id}.jsonl" + + @property + def path(self) -> Path: + return self._path + + def write( + self, + *, + procedure_id: str, + step: str, + persona: str, + inputs_hash: str, + action: str, + output_ref: Optional[str] = None, + escalated: bool = False, + operator_review_state: str = "n/a", + decision_criteria: Optional[dict[str, Any]] = None, + notes: str = "", + ) -> AuditRow: + row = AuditRow( + ts=datetime.now(timezone.utc), + property_id=self.property_id, + procedure_id=procedure_id, + step=step, + persona=persona, + inputs_hash=inputs_hash, + action=action, + output_ref=output_ref, + escalated_bool=escalated, + operator_review_state=operator_review_state, # type: ignore[arg-type] + decision_criteria=decision_criteria or {}, + notes=notes, + ) + # `mode_json` style ordered dict so on-disk fields read in audit-row + # order (BigQuery-friendly). + line = row.model_dump(mode="json") + with self._path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(line, separators=(",", ":"))) + fh.write("\n") + return row + + def read_all(self) -> list[dict[str, Any]]: + """Helper for tests: read back what we've written.""" + if not self._path.is_file(): + return [] + rows = [] + for line in self._path.read_text(encoding="utf-8").splitlines(): + if line.strip(): + rows.append(json.loads(line)) + return rows diff --git a/hermes_agent/loops/classifier.py b/hermes_agent/loops/classifier.py new file mode 100644 index 000000000000..626d8e038f5b --- /dev/null +++ b/hermes_agent/loops/classifier.py @@ -0,0 +1,123 @@ +"""P-01 — Inbound tenant comm intake. + +Classifies an inbound message into exactly one of the eight intents +defined in SOP P-01. The SOP is a deterministic first-match-wins rule +list; we delegate the matching to the model with the SOP itself in cache, +because keyword matching alone fails on polite/indirect tenant phrasing. + +The model returns a JSON object matching `Classification`. We validate +shape; if validation fails we fall back to `intent=unclassified` so the +loop continues and P-09 picks up the escalation downstream. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from .llm_client import LlmClient +from .schemas import Classification, InboundMessage + +logger = logging.getLogger(__name__) + + +P01_INSTRUCTION = """\ +You are the `property-orchestrator` persona. Apply procedure **P-01 — +Inbound tenant comm intake** from the SOP above. + +Task: classify the incoming tenant message into exactly ONE intent using +the SOP's first-match-wins decision rules: + + 1. maintenance — habitability/repair issues + 2. payment — rent, balance, autopay, receipts + 3. lease_change — renew, extend, terminate, move-out, sublease, alter + 4. notice_required — formal notice (e.g. intent to vacate) + 5. complaint_neighbor — noise, parking, smoke (cannabis context), neighbors + 6. admin — COI, keys, parking permits, contact updates + 7. legal — attorney, court, ADA, fair housing, statute citations + 8. unclassified — none of the above + +Also: identify the tenant if possible. The company state and tenant +records are in the system context above. Use the sender email or body +content to match a tenant; produce a short slug (e.g. "beautiful-minds-a-101") +if confident, else null. + +Respond with ONLY a JSON object, no prose, no code fences: + +{ + "intent": "", + "tenant_slug": "", + "confidence": , + "rationale": "" +} +""" + + +def classify( + message: InboundMessage, + *, + sop_text: str, + company_context: str, + llm: LlmClient, +) -> Classification: + """Run P-01 classification on a single inbound message. + + Args: + message: parsed inbox message. + sop_text: full SOP markdown — cached system block. + company_context: rendered per-property context — cached system block. + llm: shared LLM client (one instance reused across the inbox so + prompt cache hits across messages). + + Returns: + Classification. On validation failure, falls back to + `intent=unclassified` and surfaces the parse error in the rationale. + """ + user_payload = json.dumps(_message_payload(message), separators=(",", ":")) + + parsed: dict[str, Any] + try: + parsed, _call = llm.call_json( + cached_context_blocks=[sop_text, company_context], + instruction=P01_INSTRUCTION, + user_payload=user_payload, + max_tokens=512, + ) + except Exception as exc: # noqa: BLE001 — never let a bad LLM call break the loop + logger.exception("P-01 LLM call failed for message %s", message.id) + return Classification( + intent="unclassified", + tenant_slug=None, + confidence=0.0, + rationale=f"classifier-error: {type(exc).__name__}: {exc!s}"[:240], + ) + + try: + return Classification(**parsed) + except Exception as exc: # noqa: BLE001 — schema mismatch + logger.warning( + "P-01 returned unparseable JSON for %s: %r — defaulting to unclassified", + message.id, + parsed, + ) + return Classification( + intent="unclassified", + tenant_slug=None, + confidence=0.0, + rationale=f"schema-error: {type(exc).__name__}: {exc!s}"[:240], + ) + + +def _message_payload(message: InboundMessage) -> dict[str, Any]: + """Trim attachments to references — never feed raw blobs to the model.""" + return { + "id": message.id, + "received_at": message.received_at.isoformat(), + "channel": message.channel, + "from": message.from_, + "to": message.to, + "subject": message.subject, + "body": message.body, + "attachment_count": len(message.attachments), + } diff --git a/hermes_agent/loops/cli.py b/hermes_agent/loops/cli.py new file mode 100644 index 000000000000..94c6feb464f5 --- /dev/null +++ b/hermes_agent/loops/cli.py @@ -0,0 +1,106 @@ +"""Fire-based CLI for UCPM-specific hermes-agent commands. + +Registered as the `hermes-ucpm` console script (see pyproject.toml). Kept +in its own entry point so it can evolve independently of the upstream +`hermes` CLI defined in `hermes_cli/main.py` — that file gets rebased +from upstream and we don't want UCPM commands sitting on top of it. + +Usage examples: + + uv run hermes-ucpm test-property-loop \\ + --inbox ./inbox \\ + --outbox ./outbox \\ + --audit ./audit-log \\ + --company-dir ../paperclip-UCPM/companies/1011-verrado-office + + uv run hermes-ucpm --help +""" + +from __future__ import annotations + +import logging +import sys +from pathlib import Path +from typing import Optional + +import fire + +from .property_test_loop import LoopRunSummary, run_loop + + +def _setup_logging(verbose: bool) -> None: + level = logging.DEBUG if verbose else logging.INFO + logging.basicConfig( + level=level, + format="%(asctime)s %(levelname)-7s %(name)s: %(message)s", + ) + + +class HermesUcpmCli: + """Top-level UCPM commands.""" + + def test_property_loop( + self, + inbox: str = "./inbox", + outbox: str = "./outbox", + audit: str = "./audit-log", + company_dir: str = "../paperclip-UCPM/companies/ucpm-default", + property_id: Optional[str] = None, + verbose: bool = False, + ) -> int: + """Run P-01 + P-02 over the inbox and write drafts + audit logs. + + Args: + inbox: directory of `*.json` inbound messages. + outbox: parent directory; drafts written to `/drafts/`. + audit: directory for `.jsonl` audit logs. + company_dir: per-property company directory under + `paperclip-UCPM/companies/`. Defaults to `ucpm-default`. + property_id: optional override for the audit `property_id` + column (defaults to the company-dir basename). + verbose: enable DEBUG logging. + + Returns: + 0 on full success, 1 if any messages were skipped (malformed + input or runtime error). Per-message failures don't crash the + loop — they're captured in the summary. + """ + _setup_logging(verbose) + + summary = run_loop( + inbox_dir=Path(inbox), + outbox_dir=Path(outbox), + audit_dir=Path(audit), + company_dir=Path(company_dir), + property_id=property_id, + ) + _print_summary(summary) + return 0 if not summary.skipped else 1 + + +def _print_summary(summary: LoopRunSummary) -> None: + print(f"Processed: {len(summary.processed)}") + for r in summary.processed: + gates = ",".join(r.gates_triggered) if r.gates_triggered else "-" + urgency = r.triage.urgency if r.triage else "-" + print( + f" {r.msg_id}: intent={r.classification.intent} " + f"urgency={urgency} gates={gates} " + f"human_attention={r.human_attention_required} " + f"-> {r.draft_path}" + ) + if summary.skipped: + print(f"Skipped: {len(summary.skipped)}") + for path, reason in summary.skipped: + print(f" {path.name}: {reason}") + print(f"LLM calls: {summary.llm_calls}") + + +def main() -> None: + """Entry point referenced from pyproject.toml [project.scripts].""" + fire.Fire(HermesUcpmCli, name="hermes-ucpm") + + +if __name__ == "__main__": # pragma: no cover + main() + sys.exit(0) diff --git a/hermes_agent/loops/drafter.py b/hermes_agent/loops/drafter.py new file mode 100644 index 000000000000..451c4b9669d5 --- /dev/null +++ b/hermes_agent/loops/drafter.py @@ -0,0 +1,167 @@ +"""Draft outbound responses for the file-based test property loop. + +This is the third LLM-call site (after P-01 classify and P-02 triage). It +produces the `drafted_action` body that goes into `outbox/drafts/.json`. + +For maintenance with `urgency in {emergency, high, normal, scheduled}` we +also draft a vendor-dispatch summary (the secondary action). The actual +vendor selection in production would query `ucpm.vendors` per P-02; in +the file loop we return a "TBD - no vendor list loaded" placeholder so the +draft is reviewable end-to-end without a vendor table. + +For non-maintenance intents we draft only an acknowledgement and let +P-09 (escalation) carry the matter forward — the loop marks +`human_attention_required=true` for those. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any, Optional + +from .llm_client import LlmClient +from .schemas import ( + Classification, + DraftedAction, + InboundMessage, + Triage, + VendorDispatchDraft, +) + +logger = logging.getLogger(__name__) + + +DRAFT_INSTRUCTION = """\ +You are the `assistant-property-manager` persona. Draft an outbound email +reply to the tenant for the message below, using the SOP's tone guidance: +direct, scannable, no hedging, plain language. Do NOT cite statutes. Do +NOT promise specific dates or vendors unless given. The reply will be +queued for operator approval before sending — except the maintenance +acknowledgement template (P-02 pre-approved auto-send class), which can +be sent automatically. + +Inputs you will receive in the user payload: + - message: the inbound tenant comm. + - classification: the P-01 result. + - triage: the P-02 result if maintenance, else null. + +Respond with ONLY a JSON object, no prose: + +{ + "subject": "...", + "body": "...", + "template_id": "", + "queued_for_approval": , + "vendor_summary": "" +} + +Rules for `queued_for_approval`: + - intent=maintenance → false (ack is in the pre-approved auto-send class). + - any other intent → true. +""" + + +def draft_reply( + message: InboundMessage, + classification: Classification, + triage: Optional[Triage], + *, + sop_text: str, + company_context: str, + llm: LlmClient, +) -> tuple[DraftedAction, Optional[VendorDispatchDraft]]: + payload = { + "message": { + "from": message.from_, + "subject": message.subject, + "body": message.body, + }, + "classification": classification.model_dump(), + "triage": triage.model_dump() if triage else None, + } + user_payload = json.dumps(payload, separators=(",", ":")) + + parsed: dict[str, Any] + try: + parsed, _call = llm.call_json( + cached_context_blocks=[sop_text, company_context], + instruction=DRAFT_INSTRUCTION, + user_payload=user_payload, + max_tokens=1024, + ) + except Exception as exc: # noqa: BLE001 + logger.exception("Draft LLM call failed for message %s", message.id) + return _safe_fallback(message, classification, triage, error=str(exc)) + + try: + action = DraftedAction( + type="email_reply_to_tenant", + subject=parsed.get("subject"), + body=parsed.get("body", ""), + queued_for_approval=bool( + parsed.get( + "queued_for_approval", + classification.intent != "maintenance", + ) + ), + template_id=parsed.get("template_id"), + ) + except Exception as exc: # noqa: BLE001 + return _safe_fallback(message, classification, triage, error=str(exc)) + + secondary: Optional[VendorDispatchDraft] = None + if classification.intent == "maintenance" and triage is not None: + vendor_summary = parsed.get("vendor_summary") or _default_vendor_summary( + message, triage + ) + secondary = VendorDispatchDraft( + vendor=f"TBD - no vendor list loaded for {triage.category} yet", + work_order_summary=vendor_summary, + queued_for_approval=True, + ) + return action, secondary + + +def _safe_fallback( + message: InboundMessage, + classification: Classification, + triage: Optional[Triage], + *, + error: str, +) -> tuple[DraftedAction, Optional[VendorDispatchDraft]]: + """If the drafting call fails, produce a minimal but valid envelope so + the operator still sees something actionable. + """ + body = ( + f"[auto-fallback draft — drafting LLM call failed: {error}]\n\n" + f"Tenant message: {message.subject or '(no subject)'}\n" + f"Classified as: {classification.intent}\n" + "Operator: please draft a reply manually." + ) + action = DraftedAction( + type="email_reply_to_tenant", + subject=f"Re: {message.subject}" if message.subject else "Re: your message", + body=body, + queued_for_approval=True, # never auto-send a fallback + template_id="ack_unclassified", + ) + secondary: Optional[VendorDispatchDraft] = None + if classification.intent == "maintenance" and triage is not None: + secondary = VendorDispatchDraft( + vendor=f"TBD - no vendor list loaded for {triage.category} yet", + work_order_summary=_default_vendor_summary(message, triage), + queued_for_approval=True, + ) + return action, secondary + + +def _default_vendor_summary(message: InboundMessage, triage: Triage) -> str: + return ( + f"{triage.urgency.upper()} {triage.category} issue reported by tenant. " + f"Subject: {message.subject!r}. " + f"Operator review pending; full body in source comm." + ) diff --git a/hermes_agent/loops/llm_client.py b/hermes_agent/loops/llm_client.py new file mode 100644 index 000000000000..770a92287f3f --- /dev/null +++ b/hermes_agent/loops/llm_client.py @@ -0,0 +1,229 @@ +"""Minimal Anthropic client wrapper for the UCPM file-based test loop. + +Why a fresh client and not `agent/anthropic_adapter.py`? + The upstream adapter is a full OpenAI-style ↔ Anthropic-Messages translator + designed for the multi-turn agent runtime. It pulls in a large surface + (extended thinking, tool-use, OAuth, Bedrock, credential pools) that we + don't need for a deterministic, file-driven SOP procedure call. The right + long-term home for shared LLM helpers is a future `hermes_agent/llm/` + module — for now this stays small and focused on what P-01 + P-02 need: + 1. Stable JSON output for classification/triage decisions. + 2. Prompt caching for the SOP + company context (these are large and + reused across every message in the inbox; we should not re-pay for + them per call). + 3. A test seam so pytest mocks the Anthropic SDK without touching env. + +Default model: `claude-sonnet-4-6` (per project memory). Override via +`ANTHROPIC_MODEL` env var or the `model` kwarg on each call. + +Secrets: `ANTHROPIC_API_KEY` is read from the environment. In production +this is supplied by Doppler — never `.env`. +""" + +from __future__ import annotations + +import json +import logging +import os +from dataclasses import dataclass +from typing import Any, Optional, Protocol + +logger = logging.getLogger(__name__) + +DEFAULT_MODEL = "claude-sonnet-4-6" +DEFAULT_MAX_TOKENS = 1024 + + +class AnthropicLike(Protocol): + """Subset of the Anthropic SDK we depend on. Lets tests pass in fakes.""" + + @property + def messages(self) -> Any: ... # noqa: D401 — protocol stub + + +@dataclass +class LlmCall: + """One call's worth of state — useful for tests/debugging.""" + + system_blocks: list[dict[str, Any]] + user_text: str + response_text: str + model: str + usage: dict[str, Any] + + +class LlmClient: + """Thin wrapper around Anthropic Messages API with prompt caching. + + System prompt is structured as multiple blocks so we can place + `cache_control: {"type": "ephemeral"}` on the long, stable context + (SOP + company state). The short instruction block at the end is + *not* cached — it changes per call type (classify vs triage vs draft). + """ + + def __init__( + self, + client: Optional[AnthropicLike] = None, + model: str = DEFAULT_MODEL, + api_key: Optional[str] = None, + ): + self.model = model + self._call_count = 0 + if client is not None: + self._client = client + return + + # Lazy import — keep this module importable in CI without the SDK. + try: + import anthropic # type: ignore + except ImportError as exc: # pragma: no cover — exercised only if SDK missing + raise RuntimeError( + "anthropic SDK is not installed; install hermes-agent or " + "`uv pip install anthropic` to use LlmClient with a real backend." + ) from exc + + key = api_key or os.environ.get("ANTHROPIC_API_KEY") + if not key: + raise RuntimeError( + "ANTHROPIC_API_KEY is not set. Use Doppler to supply it: " + "`doppler run -- uv run hermes-ucpm test-property-loop ...`." + ) + self._client = anthropic.Anthropic(api_key=key) + + @property + def call_count(self) -> int: + return self._call_count + + def call_json( + self, + cached_context_blocks: list[str], + instruction: str, + user_payload: str, + *, + max_tokens: int = DEFAULT_MAX_TOKENS, + model: Optional[str] = None, + ) -> tuple[dict[str, Any], LlmCall]: + """Call the model and parse a single JSON object from its output. + + Args: + cached_context_blocks: large, stable context strings (SOP, company + state). Each becomes a system block with `cache_control` set to + ephemeral so prompt caching applies. + instruction: short, per-call-type instruction (NOT cached). + user_payload: the per-message JSON the model decides on. + max_tokens: cap on response length. + model: override the default model for this call. + """ + system_blocks: list[dict[str, Any]] = [] + for block in cached_context_blocks: + if not block: + continue + system_blocks.append( + { + "type": "text", + "text": block, + "cache_control": {"type": "ephemeral"}, + } + ) + # Final per-call instruction is intentionally uncached — it is short + # and varies between classify/triage/draft. + system_blocks.append({"type": "text", "text": instruction}) + + chosen_model = model or self.model + response = self._client.messages.create( + model=chosen_model, + max_tokens=max_tokens, + system=system_blocks, + messages=[{"role": "user", "content": user_payload}], + ) + self._call_count += 1 + + text = _extract_text(response) + parsed = _parse_json_block(text) + usage = _extract_usage(response) + + return parsed, LlmCall( + system_blocks=system_blocks, + user_text=user_payload, + response_text=text, + model=chosen_model, + usage=usage, + ) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _extract_text(response: Any) -> str: + """Pull the assistant text out of an Anthropic Messages response. + + Tolerates both real SDK responses and dict-shaped fakes used in tests. + """ + content = getattr(response, "content", None) + if content is None and isinstance(response, dict): + content = response.get("content") + if not content: + raise ValueError("Anthropic response had no content") + + # SDK form: list of content blocks each with `.text` + parts: list[str] = [] + for block in content: + text = getattr(block, "text", None) + if text is None and isinstance(block, dict): + text = block.get("text") + if text: + parts.append(text) + if not parts: + raise ValueError("Anthropic response content had no text blocks") + return "\n".join(parts) + + +def _parse_json_block(text: str) -> dict[str, Any]: + """Parse a JSON object out of the model's response. + + Allows either pure JSON or JSON inside a fenced code block. Errors here + surface as a hard failure — the loop must not silently misclassify. + """ + candidate = text.strip() + # Strip ```json ... ``` fencing if present. + if candidate.startswith("```"): + first_newline = candidate.find("\n") + if first_newline != -1: + candidate = candidate[first_newline + 1 :] + if candidate.endswith("```"): + candidate = candidate[: -len("```")].rstrip() + try: + parsed = json.loads(candidate) + except json.JSONDecodeError as exc: + raise ValueError( + f"LLM returned non-JSON response: {text[:500]!r}" + ) from exc + if not isinstance(parsed, dict): + raise ValueError( + f"LLM returned JSON of type {type(parsed).__name__}, expected object" + ) + return parsed + + +def _extract_usage(response: Any) -> dict[str, Any]: + usage = getattr(response, "usage", None) + if usage is None and isinstance(response, dict): + usage = response.get("usage") + if usage is None: + return {} + if isinstance(usage, dict): + return dict(usage) + # SDK Usage object — pull the standard fields if present. + out: dict[str, Any] = {} + for field in ( + "input_tokens", + "output_tokens", + "cache_creation_input_tokens", + "cache_read_input_tokens", + ): + value = getattr(usage, field, None) + if value is not None: + out[field] = value + return out diff --git a/hermes_agent/loops/property_test_loop.py b/hermes_agent/loops/property_test_loop.py new file mode 100644 index 000000000000..783bd202a9b9 --- /dev/null +++ b/hermes_agent/loops/property_test_loop.py @@ -0,0 +1,346 @@ +"""File-based test loop for P-01 + P-02 (CHG-0002). + +Reads inbound messages from `inbox/`, runs P-01 classification, conditionally +runs P-02 triage, drafts a response, and writes: + - `outbox/drafts/.json` — the drafted action(s) + decisions + - `audit-log/.jsonl` — one row per procedure step + +This is a local-only loop. No SMTP/IMAP, no Postgres, no external state. +Once it works against synthetic emails the agent stack is verified end-to-end +and the production wiring (SMTP/IMAP, the BigQuery `ucpm.*` ledger, the +Command Center buttons) lands later. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +from .audit import AuditWriter, compute_inputs_hash +from .classifier import classify +from .drafter import draft_reply +from .llm_client import LlmClient +from .schemas import ( + Classification, + DraftEnvelope, + DraftedAction, + InboundMessage, + Triage, + VendorDispatchDraft, +) +from .sop_loader import ( + SopBundle, + discover_inbox_messages, + load_message, + load_sop_bundle, + render_company_context_block, +) +from .triage import triage as run_triage + +logger = logging.getLogger(__name__) + + +# Intents that the SOP says cannot be auto-acknowledged and must escalate +# (legal-gate / lease-change-gate / novel-situation-gate). +_ESCALATING_INTENTS = {"lease_change", "notice_required", "legal", "unclassified"} + + +@dataclass +class LoopResult: + """One processed message's outcome.""" + + msg_id: str + draft_path: Path + audit_path: Path + classification: Classification + triage: Optional[Triage] + gates_triggered: list[str] + human_attention_required: bool + + +@dataclass +class LoopRunSummary: + """Aggregate result of running the loop over an inbox.""" + + processed: list[LoopResult] + skipped: list[tuple[Path, str]] # (path, reason) + llm_calls: int + + +def run_loop( + *, + inbox_dir: Path, + outbox_dir: Path, + audit_dir: Path, + company_dir: Path, + llm: Optional[LlmClient] = None, + property_id: Optional[str] = None, +) -> LoopRunSummary: + """Process every message in `inbox/` and write drafts + audit logs. + + Args: + inbox_dir: directory containing `*.json` inbound messages. + outbox_dir: parent dir; drafts go to `/drafts/`. + audit_dir: parent dir for `.jsonl` audit logs. + company_dir: per-property company directory under `paperclip-UCPM/companies/`. + If the SOP is not present here, the loader walks up to + `companies/ucpm-default/SOP.md`. + llm: optional preconstructed LlmClient (tests inject a fake; real + invocations let the loop construct a real Anthropic-backed one). + property_id: stable id used in audit rows. Defaults to + `.name`. + """ + inbox_dir = inbox_dir.resolve() + outbox_dir = outbox_dir.resolve() + audit_dir = audit_dir.resolve() + company_dir = company_dir.resolve() + + drafts_dir = outbox_dir / "drafts" + drafts_dir.mkdir(parents=True, exist_ok=True) + audit_dir.mkdir(parents=True, exist_ok=True) + + bundle = load_sop_bundle(company_dir) + company_ctx = render_company_context_block(bundle) + pid = property_id or bundle.company_slug + + llm_client = llm or LlmClient() + + processed: list[LoopResult] = [] + skipped: list[tuple[Path, str]] = [] + + for msg_path in discover_inbox_messages(inbox_dir): + try: + raw = load_message(msg_path) + message = InboundMessage.model_validate(raw) + except Exception as exc: # noqa: BLE001 — bad input shouldn't kill the loop + logger.exception("Skipping malformed message %s", msg_path.name) + skipped.append((msg_path, f"malformed: {type(exc).__name__}: {exc!s}")) + continue + + try: + result = _process_one( + message=message, + bundle=bundle, + company_ctx=company_ctx, + drafts_dir=drafts_dir, + audit_dir=audit_dir, + llm=llm_client, + property_id=pid, + ) + except Exception as exc: # noqa: BLE001 + logger.exception("Loop crashed on message %s", message.id) + skipped.append((msg_path, f"runtime: {type(exc).__name__}: {exc!s}")) + continue + processed.append(result) + + return LoopRunSummary( + processed=processed, + skipped=skipped, + llm_calls=llm_client.call_count, + ) + + +def _process_one( + *, + message: InboundMessage, + bundle: SopBundle, + company_ctx: str, + drafts_dir: Path, + audit_dir: Path, + llm: LlmClient, + property_id: str, +) -> LoopResult: + """Drive a single message through P-01 → (maybe P-02) → draft → audit.""" + audit = AuditWriter(audit_dir, message.id, property_id=property_id) + + msg_payload = message.model_dump(by_alias=True, mode="json") + msg_hash = compute_inputs_hash(msg_payload) + + # ----- Audit: receipt ----- + audit.write( + procedure_id="P-01", + step="receive_inbound", + persona="property-orchestrator", + inputs_hash=msg_hash, + action="received inbound comm", + notes=f"channel={message.channel} from={message.from_}", + ) + + # ----- P-01 classify ----- + classification = classify( + message, + sop_text=bundle.sop_text, + company_context=company_ctx, + llm=llm, + ) + audit.write( + procedure_id="P-01", + step="classify_intent", + persona="property-orchestrator", + inputs_hash=msg_hash, + action=f"classified intent={classification.intent}", + decision_criteria={ + "intent": classification.intent, + "tenant_slug": classification.tenant_slug, + "confidence": classification.confidence, + "rationale": classification.rationale, + }, + ) + + # ----- P-02 triage if maintenance ----- + triage: Optional[Triage] = None + if classification.intent == "maintenance": + triage = run_triage( + message, + sop_text=bundle.sop_text, + company_context=company_ctx, + llm=llm, + ) + audit.write( + procedure_id="P-02", + step="triage_maintenance", + persona="assistant-property-manager", + inputs_hash=msg_hash, + action=f"triaged urgency={triage.urgency} category={triage.category}", + decision_criteria={ + "urgency": triage.urgency, + "category": triage.category, + "payer_default": triage.payer_default, + "estimated_cost_band": triage.estimated_cost_band, + "rationale": triage.rationale, + }, + ) + + # ----- Determine gates + human-attention before drafting ----- + gates = _gates_triggered(classification, triage) + human_attention = _needs_human_attention(classification, triage, gates) + + # ----- Draft reply (and optional vendor dispatch summary) ----- + drafted_action, vendor_secondary = draft_reply( + message, + classification, + triage, + sop_text=bundle.sop_text, + company_context=company_ctx, + llm=llm, + ) + audit.write( + procedure_id="P-01" if classification.intent != "maintenance" else "P-02", + step="draft_reply", + persona="assistant-property-manager", + inputs_hash=msg_hash, + action=f"drafted {drafted_action.type} (queued={drafted_action.queued_for_approval})", + notes=f"template={drafted_action.template_id}", + ) + if vendor_secondary is not None: + audit.write( + procedure_id="P-02", + step="draft_vendor_dispatch", + persona="assistant-property-manager", + inputs_hash=msg_hash, + action="drafted vendor dispatch (TBD vendor)", + ) + + # ----- Write the draft envelope ----- + envelope = DraftEnvelope( + source_msg_id=message.id, + classification=classification, + triage=triage, + drafted_action=drafted_action, + drafted_action_secondary=vendor_secondary, + gates_triggered=gates, + human_attention_required=human_attention, + ) + draft_path = drafts_dir / f"{message.id}.json" + draft_path.write_text( + json.dumps(envelope.model_dump(mode="json"), indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + audit.write( + procedure_id="P-09" if gates else ("P-02" if triage else "P-01"), + step="emit_envelope", + persona="property-orchestrator", + inputs_hash=msg_hash, + action="wrote draft envelope", + output_ref=str(draft_path), + escalated=human_attention, + operator_review_state="pending" if drafted_action.queued_for_approval else "n/a", + decision_criteria={"gates_triggered": gates}, + ) + + return LoopResult( + msg_id=message.id, + draft_path=draft_path, + audit_path=audit.path, + classification=classification, + triage=triage, + gates_triggered=gates, + human_attention_required=human_attention, + ) + + +# --------------------------------------------------------------------------- +# Gate evaluation (rule-based, NOT LLM-driven — these are SOP invariants). +# --------------------------------------------------------------------------- + + +def _gates_triggered( + classification: Classification, triage: Optional[Triage] +) -> list[str]: + """SOP global-invariant gates — applied as pure code, never via the LLM. + + The SOP is explicit: gates (spend, legal, lease-change, novel) are + enforced by the orchestrator. We evaluate them deterministically here so + they can never silently regress when prompts change. + """ + gates: list[str] = [] + + # Legal gate. + if classification.intent == "legal": + gates.append("legal") + + # Lease-change gate. + if classification.intent in {"lease_change", "notice_required"}: + gates.append("lease-change") + + # Novel-situation gate. + if classification.intent == "unclassified": + gates.append("novel") + + # Maintenance-specific gates. + if triage is not None: + if triage.urgency == "emergency": + gates.append("emergency-vendor-dispatch") + # Spend gate: any band > $500 is a hard halt per SOP P-02. + if triage.estimated_cost_band in {"501-2000", ">2000"}: + gates.append("spend>500") + + return gates + + +def _needs_human_attention( + classification: Classification, + triage: Optional[Triage], + gates: list[str], +) -> bool: + """Should the operator be paged or surfaced at top of digest? + + Human attention is required when: + - Any gate is tripped (legal/lease-change/novel/spend/emergency). + - The intent is in the escalating set (lease_change, notice_required, + legal, unclassified) per SOP P-01 escalation rules. + - Maintenance triage is `urgency=emergency` (operator notification + within 15 min per SOP P-02). + """ + if gates: + return True + if classification.intent in _ESCALATING_INTENTS: + return True + if triage is not None and triage.urgency == "emergency": + return True + return False diff --git a/hermes_agent/loops/sample_emails/README.md b/hermes_agent/loops/sample_emails/README.md new file mode 100644 index 000000000000..c8b1ed951bbe --- /dev/null +++ b/hermes_agent/loops/sample_emails/README.md @@ -0,0 +1,31 @@ +# Sample inbound messages for the file-based property loop (CHG-0002) + +These three messages exercise the full P-01 / P-02 path from `hermes-ucpm +test-property-loop`. They are also wired in as pytest fixtures via +`tests/ucpm/conftest.py`. + +| File | Expected P-01 intent | Expected P-02 outcome | +|----------------------------|----------------------|---------------------------------------------| +| `maintenance-hvac.json` | `maintenance` | urgency=high, category=hvac | +| `rent-question.json` | `payment` | (no triage — non-maintenance) | +| `emergency-water.json` | `maintenance` | urgency=emergency, gates=[emergency, spend] | + +Run the loop locally: + +```sh +mkdir -p inbox outbox audit-log +cp hermes_agent/loops/sample_emails/*.json inbox/ +doppler run -- uv run hermes-ucpm test-property-loop \ + --inbox ./inbox \ + --outbox ./outbox \ + --audit ./audit-log \ + --company-dir ../paperclip-UCPM/companies/ucpm-default +``` + +`outbox/drafts/.json` will hold the drafted action(s); `audit-log/.jsonl` +will hold the per-step audit rows. Set `ANTHROPIC_API_KEY` via Doppler — never +in a `.env` file. + +Note: this directory is named `sample_emails/` (not `examples/`) because +the upstream `.gitignore` excludes any `examples/` directory at any depth. +Test code resolves the path via `EXAMPLES_DIR` in `tests/ucpm/conftest.py`. diff --git a/hermes_agent/loops/sample_emails/emergency-water.json b/hermes_agent/loops/sample_emails/emergency-water.json new file mode 100644 index 000000000000..6a607082e560 --- /dev/null +++ b/hermes_agent/loops/sample_emails/emergency-water.json @@ -0,0 +1,10 @@ +{ + "id": "msg-003", + "received_at": "2026-05-05T08:12:00Z", + "channel": "email", + "from": "frontdesk@beautifulmind.example", + "to": "manager@1011verrado.example", + "subject": "URGENT - water flooding from ceiling in suite A-101", + "body": "Active water leak — water is pouring through the ceiling tile above our front desk and flooding the lobby. We have patients in the waiting area. Please send someone immediately. We have shut off the main valve we could find but it is still coming down.", + "attachments": [] +} diff --git a/hermes_agent/loops/sample_emails/maintenance-hvac.json b/hermes_agent/loops/sample_emails/maintenance-hvac.json new file mode 100644 index 000000000000..581501a41cde --- /dev/null +++ b/hermes_agent/loops/sample_emails/maintenance-hvac.json @@ -0,0 +1,10 @@ +{ + "id": "msg-001", + "received_at": "2026-05-05T15:30:00Z", + "channel": "email", + "from": "office@beautifulmind.example", + "to": "manager@1011verrado.example", + "subject": "AC not cooling in suite A-101", + "body": "Hi - the AC in our suite has been blowing warm air since this morning. Patients are uncomfortable. Can someone look at this today?", + "attachments": [] +} diff --git a/hermes_agent/loops/sample_emails/rent-question.json b/hermes_agent/loops/sample_emails/rent-question.json new file mode 100644 index 000000000000..2e4eddf33b99 --- /dev/null +++ b/hermes_agent/loops/sample_emails/rent-question.json @@ -0,0 +1,10 @@ +{ + "id": "msg-002", + "received_at": "2026-05-05T16:05:00Z", + "channel": "email", + "from": "office@beautifulmind.example", + "to": "manager@1011verrado.example", + "subject": "Question about May rent balance", + "body": "Hi, our office manager is reconciling our books and wanted to confirm the balance due for May rent. Could you send a current statement showing what we owe and any credits on account? Thanks.", + "attachments": [] +} diff --git a/hermes_agent/loops/schemas.py b/hermes_agent/loops/schemas.py new file mode 100644 index 000000000000..91034ac9616f --- /dev/null +++ b/hermes_agent/loops/schemas.py @@ -0,0 +1,142 @@ +"""Pydantic schemas for the file-based test property loop. + +Shapes match the spec for CHG-0002 (P-01 + P-02). Structures here are +deliberately a subset of the full SOP — we only model what the loop reads +and writes today. Persistent stores (BigQuery `ucpm.*` tables, the draft +queue under `outbox/pending_approval/`, etc.) come later. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Literal, Optional + +from pydantic import BaseModel, Field + + +# --------------------------------------------------------------------------- +# Inbound message +# --------------------------------------------------------------------------- + + +class InboundMessage(BaseModel): + """Shape of `inbox/.json` files.""" + + id: str + received_at: datetime + channel: Literal["email", "sms", "voice", "other"] = "email" + from_: str = Field(alias="from") + to: str + subject: str = "" + body: str + attachments: list[dict[str, Any]] = Field(default_factory=list) + + model_config = {"populate_by_name": True} + + +# --------------------------------------------------------------------------- +# P-01 classification +# --------------------------------------------------------------------------- + + +# Intents are exactly the eight defined in SOP P-01 decision criteria. +Intent = Literal[ + "maintenance", + "payment", + "lease_change", + "notice_required", + "complaint_neighbor", + "admin", + "legal", + "unclassified", +] + + +class Classification(BaseModel): + intent: Intent + tenant_slug: Optional[str] = None + confidence: float = Field(ge=0.0, le=1.0) + rationale: str = "" + + +# --------------------------------------------------------------------------- +# P-02 triage +# --------------------------------------------------------------------------- + + +Urgency = Literal["emergency", "high", "normal", "scheduled"] + + +class Triage(BaseModel): + urgency: Urgency + category: str # hvac, plumbing, electrical, pest, lock, appliance, ... + rationale: str + payer_default: Literal["landlord", "tenant", "ambiguous"] = "landlord" + estimated_cost_band: Literal["unknown", "<=500", "501-2000", ">2000"] = "unknown" + + +# --------------------------------------------------------------------------- +# Drafted action(s) +# --------------------------------------------------------------------------- + + +class DraftedAction(BaseModel): + type: str # email_reply_to_tenant, vendor_dispatch_draft, ... + subject: Optional[str] = None + body: str + queued_for_approval: bool = True + template_id: Optional[str] = None # references a fixed SOP template, if any + + +class VendorDispatchDraft(BaseModel): + type: Literal["vendor_dispatch_draft"] = "vendor_dispatch_draft" + vendor: str + work_order_summary: str + queued_for_approval: bool = True + + +# --------------------------------------------------------------------------- +# Combined draft envelope +# --------------------------------------------------------------------------- + + +class DraftEnvelope(BaseModel): + """Top-level shape written to `outbox/drafts/.json`.""" + + source_msg_id: str + classification: Classification + triage: Optional[Triage] = None + drafted_action: DraftedAction + drafted_action_secondary: Optional[VendorDispatchDraft] = None + gates_triggered: list[str] = Field(default_factory=list) + human_attention_required: bool = False + + +# --------------------------------------------------------------------------- +# Audit log row +# --------------------------------------------------------------------------- + + +class AuditRow(BaseModel): + """One JSONL row in `audit-log/.jsonl`. + + Schema mirrors `ucpm.audit_log` defined in SOP global invariants: + ts, property_id, procedure_id, step, persona, inputs_hash, + action, output_ref, escalated_bool, operator_review_state. + """ + + ts: datetime + property_id: str + procedure_id: str # P-01, P-02, ... + step: str + persona: str # property-orchestrator, assistant-property-manager, ... + inputs_hash: str + action: str # short verb-phrase: "classified intent", "drafted ack", ... + output_ref: Optional[str] = None # path or id pointing at the produced artifact + escalated_bool: bool = False + operator_review_state: Literal[ + "n/a", "pending", "approved", "edited", "rejected", "deferred" + ] = "n/a" + # Loop-only addenda (not in BQ schema) — useful for debugging the file loop: + decision_criteria: dict[str, Any] = Field(default_factory=dict) + notes: str = "" diff --git a/hermes_agent/loops/sop_loader.py b/hermes_agent/loops/sop_loader.py new file mode 100644 index 000000000000..e5563b90014a --- /dev/null +++ b/hermes_agent/loops/sop_loader.py @@ -0,0 +1,154 @@ +"""Load the canonical UCPM SOP and per-company context for the loop. + +The SOP lives in `paperclip-UCPM/companies/ucpm-default/SOP.md`. Per-property +companies under `paperclip-UCPM/companies//` may add an +`SOP.overrides.yml` and a `state.yml`. This loader keeps the access pattern +explicit and read-only — the loop never mutates company-dir contents. + +The loaded SOP text is used as a prompt-cache breakpoint when calling +Anthropic, so loading once per loop invocation is the right granularity. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Optional + +import yaml + + +@dataclass(frozen=True) +class SopBundle: + """Everything the agent reads from disk before processing messages.""" + + sop_text: str + company_slug: str + company_state: dict[str, Any] = field(default_factory=dict) + overrides: dict[str, Any] = field(default_factory=dict) + extra_context: dict[str, Any] = field(default_factory=dict) + + +def _read_text(p: Path) -> str: + return p.read_text(encoding="utf-8") + + +def _read_yaml(p: Path) -> dict[str, Any]: + raw = p.read_text(encoding="utf-8") + data = yaml.safe_load(raw) or {} + if not isinstance(data, dict): + raise ValueError(f"Expected mapping at top of {p}, got {type(data).__name__}") + return data + + +def find_default_sop(company_dir: Path) -> Path: + """Resolve the canonical SOP.md from a company-dir. + + Resolution order: + 1) `/SOP.md` if present + 2) `/../ucpm-default/SOP.md` + 3) `/../../companies/ucpm-default/SOP.md` + + Raises FileNotFoundError if none exist. + """ + direct = company_dir / "SOP.md" + if direct.is_file(): + return direct + + sibling = company_dir.parent / "ucpm-default" / "SOP.md" + if sibling.is_file(): + return sibling + + upward = company_dir.parent.parent / "companies" / "ucpm-default" / "SOP.md" + if upward.is_file(): + return upward + + raise FileNotFoundError( + f"Could not locate canonical SOP.md from company-dir={company_dir}. " + "Looked at: ./SOP.md, ../ucpm-default/SOP.md, " + "../../companies/ucpm-default/SOP.md." + ) + + +def load_sop_bundle(company_dir: Path) -> SopBundle: + """Load SOP + per-company state for a property. + + The loop tolerates a stub property folder (empty / scripts/ tenants/ only). + Anything missing is treated as "use ucpm-default". + """ + company_dir = company_dir.resolve() + sop_path = find_default_sop(company_dir) + sop_text = _read_text(sop_path) + + state: dict[str, Any] = {} + state_path = company_dir / "state.yml" + if state_path.is_file(): + state = _read_yaml(state_path) + + overrides: dict[str, Any] = {} + overrides_path = company_dir / "SOP.overrides.yml" + if overrides_path.is_file(): + overrides = _read_yaml(overrides_path) + + # Per-property tenants: any *.yml directly under tenants/ is a candidate + # tenant record. We don't parse fields here — the LLM gets the raw + # YAML as context. Empty dirs return an empty list. + tenants: list[dict[str, Any]] = [] + tenants_dir = company_dir / "tenants" + if tenants_dir.is_dir(): + for tenant_file in sorted(tenants_dir.glob("*.yml")): + try: + tenants.append(_read_yaml(tenant_file)) + except Exception: # noqa: BLE001 — best-effort, never fail the loop + continue + + extra_context = { + "company_dir": str(company_dir), + "sop_path": str(sop_path), + "tenants": tenants, + } + + return SopBundle( + sop_text=sop_text, + company_slug=company_dir.name, + company_state=state, + overrides=overrides, + extra_context=extra_context, + ) + + +def render_company_context_block(bundle: SopBundle) -> str: + """Stable string form of company-context for prompt caching. + + Order matters for cache stability — keep keys sorted and serialization + deterministic so the same bundle produces byte-identical output across + runs. + """ + payload = { + "company_slug": bundle.company_slug, + "company_state": bundle.company_state, + "overrides": bundle.overrides, + "tenants": bundle.extra_context.get("tenants", []), + } + return yaml.safe_dump(payload, sort_keys=True, default_flow_style=False) + + +def load_message(path: Path) -> dict[str, Any]: + """Load and minimally validate an `inbox/.json` file.""" + import json + + raw = path.read_text(encoding="utf-8") + data = json.loads(raw) + if not isinstance(data, dict): + raise ValueError(f"{path}: expected JSON object, got {type(data).__name__}") + for required in ("id", "from", "body"): + if required not in data: + raise ValueError(f"{path}: missing required field '{required}'") + return data + + +def discover_inbox_messages(inbox_dir: Path) -> list[Path]: + """Return JSON files in inbox/, sorted by name (stable ordering).""" + if not inbox_dir.is_dir(): + return [] + return sorted(p for p in inbox_dir.glob("*.json") if p.is_file()) diff --git a/hermes_agent/loops/triage.py b/hermes_agent/loops/triage.py new file mode 100644 index 000000000000..e1fc1898eaa9 --- /dev/null +++ b/hermes_agent/loops/triage.py @@ -0,0 +1,118 @@ +"""P-02 — Maintenance request triage. + +Classifies maintenance urgency (emergency / high / normal / scheduled), +identifies category and probable payer, and returns a structured Triage. + +Spend-gate detection and explicit gate-tripping is computed by the loop +itself once we have triage output (see `test_property_loop.py`). This +module is concerned only with what the SOP P-02 decision rules say about +the message. +""" + +from __future__ import annotations + +import json +import logging +from typing import Any + +from .llm_client import LlmClient +from .schemas import InboundMessage, Triage + +logger = logging.getLogger(__name__) + + +P02_INSTRUCTION = """\ +You are the `assistant-property-manager` persona. Apply procedure **P-02 — +Maintenance request triage** from the SOP above for the given inbound +maintenance message. + +Decide: + + urgency: one of + - emergency (life-safety: gas, smoke, fire, active flood, sewage, no + heat<40F, no AC>90F+medical, sparking, broken exterior + lock, ceiling collapse, CO) + - high (habitability: no hot water, HVAC down, fridge dead, sole + toilet inop, slow leak, broken interior lock, infestation + observed) + - normal (cosmetic / non-habitability) + - scheduled (routine, not broken) + + category: short kebab-case tag (hvac, plumbing, electrical, pest, lock, + appliance, roof, structural, other). + + payer_default: "landlord", "tenant", or "ambiguous". + - default to landlord unless the body strongly indicates tenant-caused + damage. v1 leans landlord for ambiguity. + + estimated_cost_band: "<=500", "501-2000", ">2000", or "unknown". + + rationale: one short sentence explaining the urgency call, anchored in + the specific SOP rule that matched. + +Respond with ONLY a JSON object, no prose: + +{ + "urgency": "...", + "category": "...", + "rationale": "...", + "payer_default": "...", + "estimated_cost_band": "..." +} +""" + + +def triage( + message: InboundMessage, + *, + sop_text: str, + company_context: str, + llm: LlmClient, +) -> Triage: + """Run P-02 triage. On error, default to `urgency=high` (safer than normal) + and flag in rationale so the loop escalates correctly. + """ + user_payload = json.dumps(_message_payload(message), separators=(",", ":")) + + parsed: dict[str, Any] + try: + parsed, _call = llm.call_json( + cached_context_blocks=[sop_text, company_context], + instruction=P02_INSTRUCTION, + user_payload=user_payload, + max_tokens=512, + ) + except Exception as exc: # noqa: BLE001 + logger.exception("P-02 LLM call failed for message %s", message.id) + return Triage( + urgency="high", + category="other", + rationale=f"triage-error fallback: {type(exc).__name__}: {exc!s}"[:240], + payer_default="ambiguous", + estimated_cost_band="unknown", + ) + + try: + return Triage(**parsed) + except Exception as exc: # noqa: BLE001 + logger.warning( + "P-02 returned unparseable JSON for %s: %r — defaulting to high/other", + message.id, + parsed, + ) + return Triage( + urgency="high", + category="other", + rationale=f"schema-error fallback: {type(exc).__name__}: {exc!s}"[:240], + payer_default="ambiguous", + estimated_cost_band="unknown", + ) + + +def _message_payload(message: InboundMessage) -> dict[str, Any]: + return { + "id": message.id, + "from": message.from_, + "subject": message.subject, + "body": message.body, + } diff --git a/pyproject.toml b/pyproject.toml index b5de3d69f6f1..9e9cf46d8927 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -133,6 +133,9 @@ all = [ hermes = "hermes_cli.main:main" hermes-agent = "run_agent:main" hermes-acp = "acp_adapter.entry:main" +# UCPM property-management entry point — kept separate from the upstream +# `hermes` CLI so it survives upstream rebases without merge conflicts. +hermes-ucpm = "hermes_agent.loops.cli:main" [tool.setuptools] py-modules = ["run_agent", "model_tools", "toolsets", "batch_runner", "trajectory_compressor", "toolset_distributions", "cli", "hermes_constants", "hermes_state", "hermes_time", "hermes_logging", "rl_cli", "utils"] @@ -142,7 +145,7 @@ hermes_cli = ["web_dist/**/*"] gateway = ["assets/**/*"] [tool.setuptools.packages.find] -include = ["agent", "agent.*", "tools", "tools.*", "hermes_cli", "gateway", "gateway.*", "tui_gateway", "tui_gateway.*", "cron", "acp_adapter", "plugins", "plugins.*"] +include = ["agent", "agent.*", "tools", "tools.*", "hermes_cli", "gateway", "gateway.*", "tui_gateway", "tui_gateway.*", "cron", "acp_adapter", "plugins", "plugins.*", "hermes_agent", "hermes_agent.*"] [tool.pytest.ini_options] testpaths = ["tests"] diff --git a/tests/ucpm/__init__.py b/tests/ucpm/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/ucpm/conftest.py b/tests/ucpm/conftest.py new file mode 100644 index 000000000000..64899a3b6dae --- /dev/null +++ b/tests/ucpm/conftest.py @@ -0,0 +1,262 @@ +"""Test fixtures for the UCPM file-based property loop. + +Provides: + - `fake_llm` — drop-in `LlmClient` replacement that returns + pre-recorded JSON responses keyed by call type. + - `tiny_sop_text` — a minimal SOP excerpt sufficient for tests. + - `company_dir` — a tmp company directory with the tiny SOP at + `companies/ucpm-default/SOP.md`. + - `example_message_paths` — pre-built inbox containing the three + example messages from `examples/test-emails/`. +""" + +from __future__ import annotations + +import json +import shutil +from pathlib import Path +from typing import Any, Optional + +import pytest + +from hermes_agent.loops.llm_client import LlmClient + + +REPO_ROOT = Path(__file__).resolve().parents[2] +EXAMPLES_DIR = REPO_ROOT / "hermes_agent" / "loops" / "sample_emails" + + +# --------------------------------------------------------------------------- +# Fake LLM +# --------------------------------------------------------------------------- + + +class _FakeMessages: + def __init__(self, scripted: dict[str, list[dict[str, Any]]]): + self._scripted = scripted + self._cursor = {k: 0 for k in scripted} + self.calls: list[dict[str, Any]] = [] + + def create(self, *, model, max_tokens, system, messages): # noqa: D401, ANN001 + instruction = "" + for block in system: + if not block.get("cache_control"): + instruction = block["text"] + break + kind = _classify_call(instruction) + responses = self._scripted.get(kind) or self._scripted.get("default") or [] + if not responses: + raise AssertionError( + f"FakeAnthropic has no scripted response for kind={kind!r}; " + f"add one in the test." + ) + idx = min(self._cursor[kind], len(responses) - 1) + self._cursor[kind] += 1 + scripted = responses[idx] + self.calls.append( + { + "kind": kind, + "model": model, + "max_tokens": max_tokens, + "user_payload": messages[0]["content"] if messages else "", + } + ) + return _FakeResponse(text=json.dumps(scripted)) + + +class _FakeAnthropic: + def __init__(self, scripted: dict[str, list[dict[str, Any]]]): + self.messages = _FakeMessages(scripted) + + +class _FakeResponse: + def __init__(self, text: str): + self.content = [_FakeBlock(text)] + self.usage = {"input_tokens": 10, "output_tokens": 5} + + +class _FakeBlock: + def __init__(self, text: str): + self.text = text + + +def _classify_call(instruction: str) -> str: + """Disambiguate call sites by signature phrases in their instruction text. + + Order matters — the drafter instruction also mentions P-01/P-02 (because + it consumes their output), so we match on the most specific signature + first. Whitespace is collapsed before matching so multi-line prompt + headings (`Draft an outbound email\\nreply`) still hit. + """ + flat = " ".join(instruction.split()) + if "Draft an outbound email reply" in flat: + return "draft" + if "Inbound tenant comm intake" in flat: + return "classify" + if "Maintenance request triage" in flat: + return "triage" + return "default" + + +def make_fake_llm(scripted: dict[str, list[dict[str, Any]]]) -> LlmClient: + """Build an `LlmClient` whose underlying SDK is a deterministic fake.""" + fake = _FakeAnthropic(scripted) + return LlmClient(client=fake, model="claude-sonnet-4-6-fake") + + +# --------------------------------------------------------------------------- +# Default scripted responses keyed to the three example messages. +# --------------------------------------------------------------------------- + + +def default_scripts_for_examples() -> dict[str, list[dict[str, Any]]]: + """Match the three fixture emails by order of inbox traversal. + + Inbox is sorted by filename: + emergency-water.json (msg-003) + maintenance-hvac.json (msg-001) + rent-question.json (msg-002) + + Each call list maps 1:1 to that order. The drafter is invoked once per + message regardless of intent. + """ + return { + "classify": [ + # emergency-water.json -> maintenance + { + "intent": "maintenance", + "tenant_slug": "beautiful-minds-a-101", + "confidence": 0.99, + "rationale": "Active flood / water leak — clear maintenance keyword set.", + }, + # maintenance-hvac.json -> maintenance + { + "intent": "maintenance", + "tenant_slug": "beautiful-minds-a-101", + "confidence": 0.95, + "rationale": "AC not cooling — mentions broken/won't cool.", + }, + # rent-question.json -> payment + { + "intent": "payment", + "tenant_slug": "beautiful-minds-a-101", + "confidence": 0.92, + "rationale": "Asks for current statement and balance due.", + }, + ], + "triage": [ + # emergency-water.json + { + "urgency": "emergency", + "category": "plumbing", + "rationale": "Active water flow / flood — emergency tier per SOP.", + "payer_default": "landlord", + "estimated_cost_band": "501-2000", + }, + # maintenance-hvac.json + { + "urgency": "high", + "category": "hvac", + "rationale": "AC down during business hours — habitability impact.", + "payer_default": "landlord", + "estimated_cost_band": "<=500", + }, + # rent-question.json — never reaches triage (intent=payment), + # but include a stub in case ordering shifts during refactors. + { + "urgency": "normal", + "category": "other", + "rationale": "n/a — should not be invoked for payment intent.", + "payer_default": "ambiguous", + "estimated_cost_band": "unknown", + }, + ], + "draft": [ + # emergency-water.json + { + "subject": "Re: URGENT - water flooding from ceiling in suite A-101", + "body": "We've received your report and are dispatching an emergency plumber now. WO-XXXX. Expect contact within the hour.", + "template_id": "ack_maintenance", + "queued_for_approval": False, + "vendor_summary": "EMERGENCY plumbing — active flood through ceiling, A-101 lobby. Mitigate first.", + }, + # maintenance-hvac.json + { + "subject": "Re: AC not cooling in suite A-101", + "body": "Thanks for letting us know. We've opened WO-YYYY and will dispatch an HVAC tech today; expect a scheduling note within 4 hours.", + "template_id": "ack_maintenance", + "queued_for_approval": False, + "vendor_summary": "HIGH hvac — AC blowing warm air, suite A-101 medical office, business hours.", + }, + # rent-question.json + { + "subject": "Re: Question about May rent balance", + "body": "Thanks — I'll pull a current statement and send it over shortly. Queued for our property accountant.", + "template_id": "ack_payment_question", + "queued_for_approval": True, + "vendor_summary": "", + }, + ], + } + + +# --------------------------------------------------------------------------- +# Pytest fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def fake_llm() -> LlmClient: + return make_fake_llm(default_scripts_for_examples()) + + +@pytest.fixture +def tiny_sop_text() -> str: + return ( + "# UCPM SOP (test fixture)\n" + "## P-01 — classify intent into one of: maintenance, payment, ...\n" + "## P-02 — triage maintenance urgency: emergency, high, normal, scheduled.\n" + ) + + +@pytest.fixture +def company_dir(tmp_path: Path, tiny_sop_text: str) -> Path: + # Mimic the canonical layout: /companies/ucpm-default/SOP.md and a + # per-property dir at /companies//. + companies = tmp_path / "companies" + default = companies / "ucpm-default" + default.mkdir(parents=True) + (default / "SOP.md").write_text(tiny_sop_text, encoding="utf-8") + + prop = companies / "1011-verrado-office" + prop.mkdir() + (prop / "state.yml").write_text("property_id: 1011-verrado\n", encoding="utf-8") + return prop + + +@pytest.fixture +def inbox_dir(tmp_path: Path) -> Path: + inbox = tmp_path / "inbox" + inbox.mkdir() + for src in EXAMPLES_DIR.glob("*.json"): + shutil.copy(src, inbox / src.name) + return inbox + + +@pytest.fixture +def outbox_dir(tmp_path: Path) -> Path: + outbox = tmp_path / "outbox" + outbox.mkdir() + return outbox + + +@pytest.fixture +def audit_dir(tmp_path: Path) -> Path: + audit = tmp_path / "audit-log" + audit.mkdir() + return audit + + +@pytest.fixture +def example_message_paths() -> dict[str, Path]: + return {p.stem: p for p in EXAMPLES_DIR.glob("*.json")} diff --git a/tests/ucpm/test_classifier.py b/tests/ucpm/test_classifier.py new file mode 100644 index 000000000000..644ddd3754a1 --- /dev/null +++ b/tests/ucpm/test_classifier.py @@ -0,0 +1,150 @@ +"""P-01 classifier unit tests.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +from hermes_agent.loops.classifier import classify +from hermes_agent.loops.schemas import InboundMessage + +from tests.ucpm.conftest import make_fake_llm + + +def _msg(**overrides) -> InboundMessage: + base = { + "id": "msg-test", + "received_at": datetime(2026, 5, 5, 15, 30, tzinfo=timezone.utc), + "channel": "email", + "from": "tenant@example.com", + "to": "manager@example.com", + "subject": "Test", + "body": "Test", + "attachments": [], + } + base.update(overrides) + return InboundMessage.model_validate(base) + + +def test_maintenance_classification(tiny_sop_text): + llm = make_fake_llm( + { + "classify": [ + { + "intent": "maintenance", + "tenant_slug": "beautiful-minds-a-101", + "confidence": 0.95, + "rationale": "AC failure keyword.", + } + ] + } + ) + result = classify( + _msg(subject="AC not cooling", body="AC blowing warm air"), + sop_text=tiny_sop_text, + company_context="company: test\n", + llm=llm, + ) + assert result.intent == "maintenance" + assert result.tenant_slug == "beautiful-minds-a-101" + assert result.confidence == 0.95 + + +def test_payment_classification(tiny_sop_text): + llm = make_fake_llm( + { + "classify": [ + { + "intent": "payment", + "tenant_slug": None, + "confidence": 0.9, + "rationale": "Asks about balance.", + } + ] + } + ) + result = classify( + _msg(subject="balance question", body="What's my May balance?"), + sop_text=tiny_sop_text, + company_context="company: test\n", + llm=llm, + ) + assert result.intent == "payment" + assert result.tenant_slug is None + + +def test_legal_classification(tiny_sop_text): + llm = make_fake_llm( + { + "classify": [ + { + "intent": "legal", + "tenant_slug": None, + "confidence": 0.99, + "rationale": "Tenant cited an attorney.", + } + ] + } + ) + result = classify( + _msg(body="My attorney will be in touch about ADA compliance."), + sop_text=tiny_sop_text, + company_context="company: test\n", + llm=llm, + ) + assert result.intent == "legal" + + +def test_classifier_falls_back_to_unclassified_on_bad_json(tiny_sop_text): + """If the LLM returns malformed JSON the classifier must NOT crash — + it must return `unclassified` so the loop's P-09 path picks it up.""" + + class _BadAnthropic: + class _Messages: + def create(self, **kwargs): + # Return raw text that isn't JSON. + class _Block: + text = "I'm not JSON, sorry." + + class _Resp: + content = [_Block()] + usage = {} + + return _Resp() + + messages = _Messages() + + from hermes_agent.loops.llm_client import LlmClient + + llm = LlmClient(client=_BadAnthropic(), model="fake") + result = classify( + _msg(), + sop_text=tiny_sop_text, + company_context="company: test\n", + llm=llm, + ) + assert result.intent == "unclassified" + assert result.confidence == 0.0 + assert "classifier-error" in result.rationale or "schema-error" in result.rationale + + +def test_classifier_falls_back_on_invalid_intent(tiny_sop_text): + """Schema mismatch (intent not in the allowed set) must not crash.""" + llm = make_fake_llm( + { + "classify": [ + { + "intent": "spaghetti", # not in Intent literal + "tenant_slug": None, + "confidence": 0.5, + "rationale": "weird", + } + ] + } + ) + result = classify( + _msg(), + sop_text=tiny_sop_text, + company_context="company: test\n", + llm=llm, + ) + assert result.intent == "unclassified" diff --git a/tests/ucpm/test_llm_client.py b/tests/ucpm/test_llm_client.py new file mode 100644 index 000000000000..1e040f524b49 --- /dev/null +++ b/tests/ucpm/test_llm_client.py @@ -0,0 +1,97 @@ +"""LlmClient unit tests — prompt caching block layout + JSON parsing.""" + +from __future__ import annotations + +import json + +import pytest + +from hermes_agent.loops.llm_client import LlmClient + + +class _RecordingMessages: + def __init__(self, response_text): + self.response_text = response_text + self.last_kwargs = None + + def create(self, **kwargs): + self.last_kwargs = kwargs + block = type("Block", (), {"text": self.response_text})() + resp = type("Resp", (), {"content": [block], "usage": {}})() + return resp + + +class _RecordingAnthropic: + def __init__(self, response_text): + self.messages = _RecordingMessages(response_text) + + +def test_call_json_marks_long_context_blocks_as_cached(): + fake = _RecordingAnthropic(json.dumps({"intent": "maintenance"})) + llm = LlmClient(client=fake, model="m") + + parsed, call = llm.call_json( + cached_context_blocks=["LONG SOP TEXT", "COMPANY CONTEXT"], + instruction="Short instruction", + user_payload="hello", + ) + + assert parsed == {"intent": "maintenance"} + kwargs = fake.messages.last_kwargs + assert kwargs["model"] == "m" + + blocks = kwargs["system"] + # First two blocks are cached, last is not. + assert blocks[0]["text"] == "LONG SOP TEXT" + assert blocks[0]["cache_control"] == {"type": "ephemeral"} + assert blocks[1]["text"] == "COMPANY CONTEXT" + assert blocks[1]["cache_control"] == {"type": "ephemeral"} + assert blocks[2]["text"] == "Short instruction" + assert "cache_control" not in blocks[2] + + +def test_call_json_increments_call_count(): + fake = _RecordingAnthropic(json.dumps({"x": 1})) + llm = LlmClient(client=fake, model="m") + assert llm.call_count == 0 + llm.call_json(cached_context_blocks=["a"], instruction="i", user_payload="u") + llm.call_json(cached_context_blocks=["a"], instruction="i", user_payload="u") + assert llm.call_count == 2 + + +def test_call_json_strips_code_fences(): + fenced = "```json\n{\"intent\": \"payment\"}\n```" + fake = _RecordingAnthropic(fenced) + llm = LlmClient(client=fake, model="m") + parsed, _ = llm.call_json( + cached_context_blocks=["a"], instruction="i", user_payload="u" + ) + assert parsed == {"intent": "payment"} + + +def test_call_json_raises_on_non_json(): + fake = _RecordingAnthropic("definitely not json") + llm = LlmClient(client=fake, model="m") + with pytest.raises(ValueError, match="non-JSON"): + llm.call_json(cached_context_blocks=["a"], instruction="i", user_payload="u") + + +def test_call_json_raises_on_non_object_json(): + fake = _RecordingAnthropic("[1, 2, 3]") + llm = LlmClient(client=fake, model="m") + with pytest.raises(ValueError, match="expected object"): + llm.call_json(cached_context_blocks=["a"], instruction="i", user_payload="u") + + +def test_empty_context_blocks_are_dropped(): + fake = _RecordingAnthropic(json.dumps({"ok": True})) + llm = LlmClient(client=fake, model="m") + llm.call_json( + cached_context_blocks=["", "real block"], + instruction="i", + user_payload="u", + ) + blocks = fake.messages.last_kwargs["system"] + # Empty block should NOT be sent. We expect: real-block (cached) + instruction. + texts = [b["text"] for b in blocks] + assert texts == ["real block", "i"] diff --git a/tests/ucpm/test_loop_e2e.py b/tests/ucpm/test_loop_e2e.py new file mode 100644 index 000000000000..8731eaf6ccdb --- /dev/null +++ b/tests/ucpm/test_loop_e2e.py @@ -0,0 +1,201 @@ +"""End-to-end test of the file-based property loop on the three example +messages from `examples/test-emails/`. + +Exercises: + - Inbox traversal + JSON parse. + - P-01 classification (3 messages). + - P-02 triage (only the 2 maintenance messages). + - Drafter (3 messages). + - Audit log JSONL emission (one file per message, multiple rows each). + - Gate evaluation (legal / lease / spend / emergency / novel). + - Human-attention bool consistency with gates + emergency triage. + +LLM is mocked via the shared `fake_llm` fixture — no API calls. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from hermes_agent.loops.audit import AuditWriter +from hermes_agent.loops.property_test_loop import run_loop + + +def test_loop_processes_three_examples_end_to_end( + inbox_dir, outbox_dir, audit_dir, company_dir, fake_llm +): + summary = run_loop( + inbox_dir=inbox_dir, + outbox_dir=outbox_dir, + audit_dir=audit_dir, + company_dir=company_dir, + llm=fake_llm, + ) + + # All three messages processed, no skipped. + assert len(summary.processed) == 3 + assert summary.skipped == [] + + # 3 classify + 2 triage (only maintenance) + 3 draft = 8 LLM calls. + assert summary.llm_calls == 8 + + by_id = {r.msg_id: r for r in summary.processed} + assert set(by_id) == {"msg-001", "msg-002", "msg-003"} + + # ----- msg-003: emergency water leak ----- + water = by_id["msg-003"] + assert water.classification.intent == "maintenance" + assert water.triage is not None + assert water.triage.urgency == "emergency" + assert "emergency-vendor-dispatch" in water.gates_triggered + assert "spend>500" in water.gates_triggered # cost band 501-2000 + assert water.human_attention_required is True + + # ----- msg-001: AC down, high but not emergency ----- + hvac = by_id["msg-001"] + assert hvac.classification.intent == "maintenance" + assert hvac.triage is not None + assert hvac.triage.urgency == "high" + assert hvac.triage.category == "hvac" + assert hvac.gates_triggered == [] + assert hvac.human_attention_required is False + + # ----- msg-002: rent question ----- + rent = by_id["msg-002"] + assert rent.classification.intent == "payment" + assert rent.triage is None + assert rent.gates_triggered == [] + assert rent.human_attention_required is False + + +def test_loop_writes_draft_envelope_with_correct_shape( + inbox_dir, outbox_dir, audit_dir, company_dir, fake_llm +): + run_loop( + inbox_dir=inbox_dir, + outbox_dir=outbox_dir, + audit_dir=audit_dir, + company_dir=company_dir, + llm=fake_llm, + ) + + drafts_dir = outbox_dir / "drafts" + written = sorted(p.name for p in drafts_dir.glob("*.json")) + assert written == ["msg-001.json", "msg-002.json", "msg-003.json"] + + water = json.loads((drafts_dir / "msg-003.json").read_text(encoding="utf-8")) + assert water["source_msg_id"] == "msg-003" + assert water["classification"]["intent"] == "maintenance" + assert water["triage"]["urgency"] == "emergency" + assert "emergency-vendor-dispatch" in water["gates_triggered"] + assert water["human_attention_required"] is True + assert water["drafted_action"]["type"] == "email_reply_to_tenant" + assert water["drafted_action_secondary"]["type"] == "vendor_dispatch_draft" + # No vendor list loaded yet, so we expect a TBD placeholder. + assert "TBD" in water["drafted_action_secondary"]["vendor"] + + rent = json.loads((drafts_dir / "msg-002.json").read_text(encoding="utf-8")) + assert rent["classification"]["intent"] == "payment" + assert rent["triage"] is None + assert rent["drafted_action"]["queued_for_approval"] is True + assert rent["drafted_action_secondary"] is None + + +def test_loop_writes_audit_log_per_message( + inbox_dir, outbox_dir, audit_dir, company_dir, fake_llm +): + run_loop( + inbox_dir=inbox_dir, + outbox_dir=outbox_dir, + audit_dir=audit_dir, + company_dir=company_dir, + llm=fake_llm, + ) + + written = sorted(p.name for p in audit_dir.glob("*.jsonl")) + assert written == ["msg-001.jsonl", "msg-002.jsonl", "msg-003.jsonl"] + + # Maintenance flow: receive, classify(P-01), triage(P-02), draft, vendor, emit. + hvac_rows = [ + json.loads(line) + for line in (audit_dir / "msg-001.jsonl").read_text(encoding="utf-8").splitlines() + if line.strip() + ] + procedures = [r["procedure_id"] for r in hvac_rows] + steps = [r["step"] for r in hvac_rows] + assert "P-01" in procedures + assert "P-02" in procedures + assert "classify_intent" in steps + assert "triage_maintenance" in steps + assert "draft_reply" in steps + assert "draft_vendor_dispatch" in steps + assert "emit_envelope" in steps + + # Every row carries the SOP audit-log columns. + required_columns = { + "ts", + "property_id", + "procedure_id", + "step", + "persona", + "inputs_hash", + "action", + "escalated_bool", + "operator_review_state", + } + for row in hvac_rows: + assert required_columns.issubset(row.keys()), ( + f"missing audit columns: {required_columns - row.keys()}" + ) + # Property id pulled from company-dir basename. + assert hvac_rows[0]["property_id"] == "1011-verrado-office" + + # Non-maintenance flow has NO P-02 row. + rent_rows = [ + json.loads(line) + for line in (audit_dir / "msg-002.jsonl").read_text(encoding="utf-8").splitlines() + if line.strip() + ] + assert all(r["procedure_id"] != "P-02" for r in rent_rows) + + # Emergency flow escalates the final emit row. + water_rows = [ + json.loads(line) + for line in (audit_dir / "msg-003.jsonl").read_text(encoding="utf-8").splitlines() + if line.strip() + ] + emit_row = next(r for r in water_rows if r["step"] == "emit_envelope") + assert emit_row["escalated_bool"] is True + + +def test_loop_skips_malformed_inbox_entries( + inbox_dir, outbox_dir, audit_dir, company_dir, fake_llm +): + bogus = inbox_dir / "bogus.json" + bogus.write_text("{not valid json", encoding="utf-8") + + summary = run_loop( + inbox_dir=inbox_dir, + outbox_dir=outbox_dir, + audit_dir=audit_dir, + company_dir=company_dir, + llm=fake_llm, + ) + + assert len(summary.processed) == 3 + assert len(summary.skipped) == 1 + skipped_path, reason = summary.skipped[0] + assert skipped_path.name == "bogus.json" + assert "malformed" in reason + + +def test_audit_writer_inputs_hash_is_stable(tmp_path: Path): + writer = AuditWriter(tmp_path, "msg-x", property_id="prop-1") + payload = {"a": 1, "b": [1, 2, 3]} + from hermes_agent.loops.audit import compute_inputs_hash + + h1 = compute_inputs_hash(payload) + h2 = compute_inputs_hash({"b": [1, 2, 3], "a": 1}) # same content, different key order + assert h1 == h2 + assert len(h1) == 64 # sha256 hex diff --git a/tests/ucpm/test_sop_loader.py b/tests/ucpm/test_sop_loader.py new file mode 100644 index 000000000000..d23d08e3b76b --- /dev/null +++ b/tests/ucpm/test_sop_loader.py @@ -0,0 +1,115 @@ +"""SOP loader unit tests — resolution paths and stub-property tolerance.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from hermes_agent.loops.sop_loader import ( + discover_inbox_messages, + find_default_sop, + load_message, + load_sop_bundle, + render_company_context_block, +) + + +def _make_layout(tmp_path: Path, *, with_property_sop: bool = False) -> Path: + companies = tmp_path / "companies" + default = companies / "ucpm-default" + default.mkdir(parents=True) + (default / "SOP.md").write_text("# default SOP\n", encoding="utf-8") + + prop = companies / "1011-verrado-office" + prop.mkdir() + if with_property_sop: + (prop / "SOP.md").write_text("# property override SOP\n", encoding="utf-8") + return prop + + +def test_find_default_sop_resolves_via_sibling_ucpm_default(tmp_path): + prop = _make_layout(tmp_path) + sop_path = find_default_sop(prop) + assert sop_path.read_text(encoding="utf-8") == "# default SOP\n" + + +def test_find_default_sop_prefers_property_sop_if_present(tmp_path): + prop = _make_layout(tmp_path, with_property_sop=True) + sop_path = find_default_sop(prop) + assert sop_path.read_text(encoding="utf-8") == "# property override SOP\n" + + +def test_find_default_sop_raises_when_missing(tmp_path): + # No companies/ucpm-default/SOP.md anywhere. + standalone = tmp_path / "standalone" + standalone.mkdir() + with pytest.raises(FileNotFoundError): + find_default_sop(standalone) + + +def test_load_sop_bundle_handles_stub_property_dir(tmp_path): + """A property dir with no state.yml / no tenants/ must still load.""" + prop = _make_layout(tmp_path) + bundle = load_sop_bundle(prop) + assert bundle.company_slug == "1011-verrado-office" + assert bundle.company_state == {} + assert bundle.overrides == {} + assert bundle.extra_context["tenants"] == [] + + +def test_load_sop_bundle_reads_state_and_overrides(tmp_path): + prop = _make_layout(tmp_path) + (prop / "state.yml").write_text( + "property_id: 1011-verrado\n" + "owner_email: matt@example.com\n", + encoding="utf-8", + ) + (prop / "SOP.overrides.yml").write_text( + "overrides:\n" + " P-04:\n" + " grace_period_days: 3\n", + encoding="utf-8", + ) + tenants = prop / "tenants" + tenants.mkdir() + (tenants / "beautiful-minds-a-101.yml").write_text( + "slug: beautiful-minds-a-101\n" + "primary_contact: office@beautifulmind.example\n", + encoding="utf-8", + ) + + bundle = load_sop_bundle(prop) + assert bundle.company_state["property_id"] == "1011-verrado" + assert bundle.overrides["overrides"]["P-04"]["grace_period_days"] == 3 + assert bundle.extra_context["tenants"][0]["slug"] == "beautiful-minds-a-101" + + +def test_render_company_context_block_is_deterministic(tmp_path): + """Cache stability requires byte-identical output across runs for same inputs.""" + prop = _make_layout(tmp_path) + bundle = load_sop_bundle(prop) + a = render_company_context_block(bundle) + b = render_company_context_block(bundle) + assert a == b + + +def test_load_message_validates_required_fields(tmp_path): + p = tmp_path / "bad.json" + p.write_text('{"id": "x"}', encoding="utf-8") # missing 'from' and 'body' + with pytest.raises(ValueError, match="missing required field"): + load_message(p) + + +def test_discover_inbox_messages_returns_sorted_json_files(tmp_path): + inbox = tmp_path / "inbox" + inbox.mkdir() + (inbox / "z.json").write_text("{}", encoding="utf-8") + (inbox / "a.json").write_text("{}", encoding="utf-8") + (inbox / "ignored.txt").write_text("nope", encoding="utf-8") + files = discover_inbox_messages(inbox) + assert [f.name for f in files] == ["a.json", "z.json"] + + +def test_discover_inbox_messages_returns_empty_for_missing_dir(tmp_path): + assert discover_inbox_messages(tmp_path / "nope") == [] diff --git a/tests/ucpm/test_triage.py b/tests/ucpm/test_triage.py new file mode 100644 index 000000000000..be5af3f4bba9 --- /dev/null +++ b/tests/ucpm/test_triage.py @@ -0,0 +1,131 @@ +"""P-02 triage unit tests.""" + +from __future__ import annotations + +from datetime import datetime, timezone + +import pytest + +from hermes_agent.loops.schemas import InboundMessage +from hermes_agent.loops.triage import triage as run_triage + +from tests.ucpm.conftest import make_fake_llm + + +def _msg(body: str, subject: str = "") -> InboundMessage: + return InboundMessage.model_validate( + { + "id": "msg-test", + "received_at": datetime(2026, 5, 5, 8, 0, tzinfo=timezone.utc), + "channel": "email", + "from": "tenant@example.com", + "to": "manager@example.com", + "subject": subject, + "body": body, + "attachments": [], + } + ) + + +@pytest.mark.parametrize( + "scripted, expected_urgency", + [ + ( + { + "urgency": "emergency", + "category": "plumbing", + "rationale": "active flood", + "payer_default": "landlord", + "estimated_cost_band": "501-2000", + }, + "emergency", + ), + ( + { + "urgency": "high", + "category": "hvac", + "rationale": "AC down, business hours", + "payer_default": "landlord", + "estimated_cost_band": "<=500", + }, + "high", + ), + ( + { + "urgency": "normal", + "category": "plumbing", + "rationale": "slow drain", + "payer_default": "landlord", + "estimated_cost_band": "<=500", + }, + "normal", + ), + ( + { + "urgency": "scheduled", + "category": "hvac", + "rationale": "filter change request", + "payer_default": "landlord", + "estimated_cost_band": "<=500", + }, + "scheduled", + ), + ], +) +def test_triage_urgency_levels(tiny_sop_text, scripted, expected_urgency): + llm = make_fake_llm({"triage": [scripted]}) + result = run_triage( + _msg("anything"), + sop_text=tiny_sop_text, + company_context="company: test\n", + llm=llm, + ) + assert result.urgency == expected_urgency + assert result.category == scripted["category"] + + +def test_triage_falls_back_on_invalid_urgency(tiny_sop_text): + """A bad urgency value must not crash the loop — must default to high + so the operator surfaces it (safer than dropping to normal silently).""" + llm = make_fake_llm( + { + "triage": [ + { + "urgency": "kinda-bad", # invalid + "category": "hvac", + "rationale": "x", + "payer_default": "landlord", + "estimated_cost_band": "<=500", + } + ] + } + ) + result = run_triage( + _msg("AC down"), + sop_text=tiny_sop_text, + company_context="company: test\n", + llm=llm, + ) + assert result.urgency == "high" + assert "schema-error" in result.rationale + + +def test_triage_falls_back_on_llm_exception(tiny_sop_text): + class _BoomAnthropic: + class _Messages: + def create(self, **kwargs): + raise RuntimeError("API exploded") + + messages = _Messages() + + from hermes_agent.loops.llm_client import LlmClient + + llm = LlmClient(client=_BoomAnthropic(), model="fake") + result = run_triage( + _msg("AC down"), + sop_text=tiny_sop_text, + company_context="company: test\n", + llm=llm, + ) + assert result.urgency == "high" + assert "triage-error" in result.rationale