#!/usr/bin/env python3
"""HW_TENANT_PATCH applier (#4750) — surgical build-time patch of Hermes'
bundled ``observability/langfuse`` plugin.

Why a Python applier instead of a .diff + ``patch``: the base image is not
guaranteed to ship the ``patch`` binary, and exact-anchor string surgery fails
LOUDLY on upstream drift (the #4060 lesson: silent non-application is this
plugin's signature failure mode). The base image is digest-pinned in
``agent/Dockerfile``, so the anchors below are deterministic until the pin is
bumped — at which point the repo test (``evals/test_langfuse_plugin_patch.py``,
which applies this patcher to the vendored upstream copy) forces a refresh.

v2 (2026-07-22): the v1 approach (post-hoc ``root_span.update_trace(...)``)
silently failed in production — langfuse SDK 4.14's ``LangfuseChain`` has NO
``update_trace`` method (verified: AttributeError), and the fail-open except
swallowed it, taking the metadata merge down with it. v2 applies everything
AT THE SOURCE instead, using APIs that verifiably exist in 4.14:

1. Tenant attribution (hunk A + B) — read the humanwork per-turn env contract
   (``HERMES_LANGFUSE_SESSION_ID`` / ``_USER_ID`` / ``_TRACE_METADATA``,
   emitted by ``agent/main.py::_hermes_env`` since #4749, pinned by
   ``evals/test_hermes_env_langfuse.py``) right after the trace metadata
   literal is built:
   - merge extra metadata into the dict BEFORE ``start_as_current_observation``
     consumes it (no post-hoc API needed);
   - REBIND the local ``session_id`` to the platform conversation id, so
     ``trace_ctx["session_id"]`` AND ``propagate_attributes(session_id=…)``
     — and therefore every child span — carry it; Hermes' internal
     ``date_time_hash`` session is preserved as ``metadata.hermes_session``;
   - pass ``user_id`` (the org id) through ``propagate_attributes`` (hunk B),
     the sanctioned v4 channel (signature verified: it accepts ``user_id``).
     Langfuse's per-user cost view then doubles as per-org cost accounting;
   - append value-bearing tags from ``HERMES_LANGFUSE_TAGS`` (#4792 —
     ``specialist:<id>``, ``source:agent``) to the upstream constant tag
     list in the same ``propagate_attributes`` call: tags are the only
     dimension the self-hosted Metrics API can both filter AND group by
     (metadata is filter-only — spike findings, spec §7).
   Failures are logged via the plugin's own ``_debug`` (visible with
   ``HERMES_LANGFUSE_DEBUG=true``), never raised — and the v1 mistake of
   putting independent steps in one try block is not repeated.

2. Cost-zero suppression (hunks C + D) — upstream's fallback branches write
   ``cost_details["total"] = float(cost.amount_usd)`` even when the estimate
   is 0. Langfuse prioritizes ingested cost over inferred cost
   (https://langfuse.com/docs/model-usage-and-cost), so an explicit 0 total
   suppresses server-side inference from the model-price table. Guard every
   such branch: only report a total when it is > 0. There are THREE writers —
   two in ``_usage_and_cost`` (hunk C, variable ``cost``) and one in
   ``on_post_llm_call``'s usage-dict path (hunk D, variable ``_cost`` — the
   branch Hermes actually hits in production, missed by v2).

3. Cost-total sum (hunk E) — when the pricing entry DOES match, the plugin
   ingests a per-type cost breakdown but never a "total". Langfuse sums
   partial usage_details into usage.total but does NOT sum cost_details, and
   ingested cost disables its price-table inference — Total cost reads $0
   despite correct per-type costs (verified live 2026-07-22). Sum the
   breakdown into "total" at ``_end_observation``, the single sink both
   cost paths flow through.

Exit codes: 0 = applied (or already applied — idempotent), 2 = anchor not
found / ambiguous (upstream drifted; refresh anchors + vendored copy).
"""
from __future__ import annotations

import sys
from pathlib import Path

MARKER = "HW_TENANT_PATCH"

# ── Hunk A: tenant env → metadata merge + session rebind ─────────────────────
# Anchor = the root-trace metadata literal in `_start_root_trace` (unique via
# its `"source": "hermes"` line). The insert runs BEFORE trace_ctx is built
# and BEFORE any span exists, so the rebound session_id flows everywhere.
METADATA_ANCHOR = '''    metadata = {
        "source": "hermes",
        "task_id": task_id,
        "platform": platform,
        "provider": provider,
        "model": model,
        "api_mode": api_mode,
    }
'''

METADATA_INSERT = METADATA_ANCHOR + '''
    # HW_TENANT_PATCH (#4750, humanwork): tenant attribution from env, applied
    # at the SOURCE — SDK 4.14's LangfuseChain has no update_trace(), so
    # post-hoc mutation is impossible. session_id is rebound so trace_ctx AND
    # propagate_attributes (children included) carry the platform conversation
    # id; Hermes' internal session id survives as metadata.hermes_session.
    # Fail-open per step, logged via _debug (HERMES_LANGFUSE_DEBUG=true).
    _hw_user_id = None
    _hw_tags = []
    try:
        _hw_session = os.environ.get("HERMES_LANGFUSE_SESSION_ID", "").strip()
        _hw_user = os.environ.get("HERMES_LANGFUSE_USER_ID", "").strip()
        _hw_meta_raw = os.environ.get("HERMES_LANGFUSE_TRACE_METADATA", "").strip()
        # HW_TENANT_PATCH (#4792): value-bearing tags (specialist:<id>,
        # source:agent) — tags are the only dimension the Metrics API can
        # both filter and group by; comma-separated on the env contract.
        _hw_tags = [
            _hw_t.strip()
            for _hw_t in os.environ.get("HERMES_LANGFUSE_TAGS", "").split(",")
            if _hw_t.strip()
        ]
        if _hw_session:
            metadata["hermes_session"] = session_id
            session_id = _hw_session
        if _hw_user:
            _hw_user_id = _hw_user
        if _hw_meta_raw:
            try:
                _hw_extra = json.loads(_hw_meta_raw)
                if isinstance(_hw_extra, dict):
                    metadata.update({str(_hw_k): _hw_v for _hw_k, _hw_v in _hw_extra.items()})
            except Exception as _hw_exc:
                _debug(f"hw tenant metadata parse skipped: {_hw_exc}")
    except Exception as _hw_exc:  # pragma: no cover - fail-open
        _debug(f"hw tenant attribution skipped: {_hw_exc}")
'''

# ── Hunk B: user_id through propagate_attributes ─────────────────────────────
PROPAGATE_ANCHOR = '''            with propagate_attributes(
                session_id=session_id or task_key,
                trace_name="Hermes turn",
                tags=["hermes", "langfuse"],
            ):
'''

PROPAGATE_REPLACEMENT = '''            with propagate_attributes(
                user_id=_hw_user_id,  # HW_TENANT_PATCH (#4750): org id, None-safe
                session_id=session_id or task_key,
                trace_name="Hermes turn",
                # HW_TENANT_PATCH (#4792): value-bearing tenant tags appended
                tags=["hermes", "langfuse"] + _hw_tags,
            ):
'''

# ── Hunk C: cost-zero guard on both fallback branches ────────────────────────
COST_ANCHOR = '''                else:
                    cost_details["total"] = float(cost.amount_usd)
            except Exception:
                cost_details["total"] = float(cost.amount_usd)
'''

COST_REPLACEMENT = '''                else:
                    # HW_TENANT_PATCH (#4750): an ingested 0 total overrides
                    # Langfuse's own price-table inference — omit instead.
                    if float(cost.amount_usd) > 0:
                        cost_details["total"] = float(cost.amount_usd)
            except Exception:
                if float(cost.amount_usd) > 0:  # HW_TENANT_PATCH (#4750)
                    cost_details["total"] = float(cost.amount_usd)
'''

# ── Hunk D: cost-zero guard on the usage-dict (post_api_request) path ────────
# `on_post_llm_call` has a SECOND, independent cost writer for calls that come
# in as a pre-built usage summary dict instead of a response object — the path
# Hermes actually takes in production. Its fallback names the variable `_cost`
# (not `cost`), so hunk C's anchors never touched it and it kept ingesting an
# explicit 0 total, which suppressed Langfuse's price-table inference exactly
# like the hunk-C branches did.
COST2_ANCHOR = '''            else:
                _cost = estimate_usage_cost(model, _cu, provider=provider, base_url=base_url, api_key="")
                if _cost.amount_usd is not None:
                    cost_details["total"] = float(_cost.amount_usd)
'''

COST2_REPLACEMENT = '''            else:
                _cost = estimate_usage_cost(model, _cu, provider=provider, base_url=base_url, api_key="")
                # HW_TENANT_PATCH (#4750): same ingested-0 suppression as hunk C.
                if _cost.amount_usd is not None and float(_cost.amount_usd) > 0:
                    cost_details["total"] = float(_cost.amount_usd)
'''

# ── Hunk E: sum per-type cost_details into a "total" at the single sink ──────
# When `get_pricing_entry` matches (the branch production actually takes for
# anthropic/claude-sonnet-4), the plugin ingests a per-type cost breakdown
# (input/output/cache_*) but never a "total". Langfuse sums partial
# usage_details into usage.total, but does NOT do the same for cost_details —
# and ingested-cost presence disables price-table inference — so the trace
# lists Total cost $0 despite correct per-type costs (verified live 2026-07-22
# via /api/public/observations: costDetails populated, calculatedTotalCost=0).
# `_end_observation` is the single sink every generation flows through
# (response-object path AND usage-dict path), so summing here covers both.
COST_TOTAL_ANCHOR = '''        if cost_details:
            update_kwargs["cost_details"] = cost_details
'''

COST_TOTAL_REPLACEMENT = '''        if cost_details:
            # HW_TENANT_PATCH (#4750): Langfuse does not sum a partial
            # per-type cost breakdown into a total (unlike usage_details),
            # and ingested cost disables its price-table inference — so a
            # breakdown without "total" reads as Total cost $0. Sum it here;
            # the > 0 guard keeps the hunk C/D zero-suppression intact.
            if "total" not in cost_details:
                try:
                    _hw_cost_total = float(sum(
                        v for v in cost_details.values()
                        if isinstance(v, (int, float))
                    ))
                    if _hw_cost_total > 0:
                        cost_details["total"] = _hw_cost_total
                except Exception as _hw_exc:
                    _debug(f"hw cost total sum skipped: {_hw_exc}")
            update_kwargs["cost_details"] = cost_details
'''

HUNKS = (
    ("metadata/session", METADATA_ANCHOR, METADATA_INSERT),
    ("propagate/user", PROPAGATE_ANCHOR, PROPAGATE_REPLACEMENT),
    ("cost", COST_ANCHOR, COST_REPLACEMENT),
    ("cost-usage-dict", COST2_ANCHOR, COST2_REPLACEMENT),
    ("cost-total-sum", COST_TOTAL_ANCHOR, COST_TOTAL_REPLACEMENT),
)


def apply(path: Path) -> int:
    source = path.read_text(encoding="utf-8")

    if MARKER in source:
        print(f"[hw-tenant-patch] already applied to {path} — no-op")
        return 0

    for name, anchor, _ in HUNKS:
        count = source.count(anchor)
        if count != 1:
            print(
                f"[hw-tenant-patch] FATAL: {name} anchor matched {count} times "
                f"(expected exactly 1) in {path}. Upstream plugin drifted — "
                f"refresh the anchors and the vendored copy in "
                f"agent/deploy/patches/.",
                file=sys.stderr,
            )
            return 2

    patched = source
    for _, anchor, replacement in HUNKS:
        patched = patched.replace(anchor, replacement, 1)

    # Belt-and-braces: the inserts rely on names that must exist in the
    # target module (imports + helpers at the insertion scope).
    for required in ("import json", "import os", "def _debug(", "propagate_attributes"):
        if required not in patched:
            print(
                f"[hw-tenant-patch] FATAL: expected '{required}' in target "
                f"module — refusing to apply.",
                file=sys.stderr,
            )
            return 2

    path.write_text(patched, encoding="utf-8")
    print(f"[hw-tenant-patch] applied to {path}")
    return 0


def main(argv: list[str]) -> int:
    if len(argv) != 2:
        print(
            "usage: apply_langfuse_tenant_patch.py <path-to-plugin-__init__.py>",
            file=sys.stderr,
        )
        return 2
    target = Path(argv[1])
    if not target.is_file():
        print(f"[hw-tenant-patch] FATAL: no such file: {target}", file=sys.stderr)
        return 2
    return apply(target)


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))