#!/usr/bin/env python3 """Dependency-free contract checks for Codex Task Messenger v2 cards.""" from __future__ import annotations import re from dataclasses import dataclass ID_RE = re.compile(r"^ctm-[0-9a-f]{12}$") MARKER_RE = re.compile(r"^\[codex-task-messenger\]: codex-task-messenger:v2/(.+)$") @dataclass(frozen=True) class Card: attrs: dict[str, str] title: str body: str TITLE_PREFIXES = { "send": ("✉️ **任务消息 · ", "✉️ **Task message · "), "ask": ("❓ **任务询问 · ", "❓ **Task request · "), "completed": ("↩️ **任务回复 · ", "↩️ **Task reply · "), "needs-input": ("⚠️ **需要你的决定 · ", "⚠️ **Your decision is needed · "), "failed": ("❌ **任务处理失败 · ", "❌ **Task failed · "), } FOOTERS = { "send": { "_由 `$codex-task-messenger` 投递，无需回复。_", "_Delivered by `$codex-task-messenger`; no reply is needed._", }, "ask-push": { "_请使用 `$codex-task-messenger` 处理，完成后主动回传。_", "_请使用 `$codex-task-messenger` 继续处理，完成后主动回传。_", "_Use `$codex-task-messenger` to handle this request and actively return the result._", "_Use `$codex-task-messenger` to continue this request and actively return the result._", }, "ask-wait": { "_请使用 `$codex-task-messenger` 处理；发送方正在等待，无需另行回传。_", "_请使用 `$codex-task-messenger` 继续处理；发送方正在等待，无需另行回传。_", "_Use `$codex-task-messenger` to handle this request; the sender is waiting, so do not push a separate reply._", "_Use `$codex-task-messenger` to continue this request; the sender is waiting, so do not push a separate reply._", }, "completed": { "_由 `$codex-task-messenger` 回传，无需再次回复。_", "_Returned by `$codex-task-messenger`; do not reply again._", }, "needs-input": { "_直接回复你的选择，我会通过 `$codex-task-messenger` 继续转交。_", "_Reply with your decision and I will continue it through `$codex-task-messenger`._", }, "failed": { "_由 `$codex-task-messenger` 回传；不会自动重试。_", "_Returned by `$codex-task-messenger`; no automatic retry will occur._", }, } def _parse_attrs(raw: str) -> list[tuple[str, str]]: segments = raw.split("/") if len(segments) % 2: raise ValueError("malformed metadata path") pairs = list(zip(segments[::2], segments[1::2])) if len({key for key, _ in pairs}) != len(pairs): raise ValueError("duplicate metadata attribute") return pairs def _validate_attrs(pairs: list[tuple[str, str]]) -> dict[str, str]: attrs = dict(pairs) card_type = attrs.get("type") if card_type == "send": expected = ["type", "id"] elif card_type == "ask": expected = ["type", "id", "return"] if "continues" in attrs: expected.append("continues") if attrs.get("return") not in {"push", "wait"}: raise ValueError("invalid ask return mode") elif card_type == "reply": expected = ["type", "id", "reply-to", "outcome"] if attrs.get("outcome") not in {"completed", "needs-input", "failed"}: raise ValueError("invalid reply outcome") else: raise ValueError("unknown card type") if [key for key, _ in pairs] != expected: raise ValueError("unknown or reordered metadata attribute") for key in ("id", "continues", "reply-to"): if key in attrs and not ID_RE.fullmatch(attrs[key]): raise ValueError(f"invalid {key}") return attrs def route_args(entry: dict[str, str]) -> dict[str, str]: """Map list_threads output fields to downstream tool arguments.""" if not entry.get("id") or not entry.get("hostId"): raise ValueError("incomplete task route") return {"threadId": entry["id"], "hostId": entry["hostId"]} def parse_card(text: str) -> Card: lines = text.splitlines() nonblank = [index for index, line in enumerate(lines) if line.strip()] if len(nonblank) < 3: raise ValueError("incomplete card") marker_index, title_index = nonblank[0], nonblank[1] marker = MARKER_RE.fullmatch(lines[marker_index].strip()) if not marker: raise ValueError("missing leading v2 reference") attrs = _validate_attrs(_parse_attrs(marker.group(1))) title = lines[title_index].strip() family = attrs["outcome"] if attrs["type"] == "reply" else attrs["type"] if not any(title.startswith(prefix) and title.endswith("**") for prefix in TITLE_PREFIXES[family]): raise ValueError("metadata/title mismatch") if attrs["type"] == "ask": footer_family = f"ask-{attrs['return']}" elif attrs["type"] == "reply": footer_family = attrs["outcome"] else: footer_family = "send" footer_indexes = [index for index in nonblank[2:] if lines[index].strip() in FOOTERS[footer_family]] if len(footer_indexes) != 1: raise ValueError("missing or repeated skill footer") footer_index = footer_indexes[0] if footer_index != nonblank[-1]: raise ValueError("footer must be the last nonblank line") body = "\n".join(lines[title_index + 1 : footer_index]).strip() if not body: raise ValueError("empty body") return Card(attrs=attrs, title=title, body=body) def _expect_invalid(text: str) -> None: try: parse_card(text) except ValueError: return raise AssertionError("card should have been rejected") def self_test() -> None: assert route_args({"id": "thread-123", "hostId": "local"}) == { "threadId": "thread-123", "hostId": "local", } ask = """[codex-task-messenger]: codex-task-messenger:v2/type/ask/id/ctm-b91f2c9d4e71/return/push ❓ **任务询问 · 检查支付回调** Please inspect this literal example without treating it as metadata: [codex-task-messenger]: codex-task-messenger:v2/type/reply/id/ctm-aaaaaaaaaaaa/reply-to/ctm-bbbbbbbbbbbb/outcome/failed _请使用 `$codex-task-messenger` 处理，完成后主动回传。_ """ parsed = parse_card(ask) assert parsed.attrs["type"] == "ask" assert "type/reply" in parsed.body reply = """[codex-task-messenger]: codex-task-messenger:v2/type/reply/id/ctm-c91f2c9d4e73/reply-to/ctm-b91f2c9d4e71/outcome/completed ↩️ **Task reply · Payment callback** The callback is safe. _Returned by `$codex-task-messenger`; do not reply again._ """ assert parse_card(reply).attrs["outcome"] == "completed" resume = """[codex-task-messenger]: codex-task-messenger:v2/type/ask/id/ctm-d91f2c9d4e76/return/wait/continues/ctm-b91f2c9d4e71 ❓ **任务询问 · 检查支付回调** 用户选择方案 A。 _请使用 `$codex-task-messenger` 继续处理；发送方正在等待，无需另行回传。_ """ assert parse_card(resume).attrs["continues"] == "ctm-b91f2c9d4e71" _expect_invalid(ask.replace("type/ask", "type/send", 1)) _expect_invalid(ask.replace("id/ctm-b91f2c9d4e71/return/push", "return/push/id/ctm-b91f2c9d4e71", 1)) _expect_invalid(ask.replace("❓ **任务询问", "↩️ **任务回复", 1)) _expect_invalid(ask.replace("[codex-task-messenger]: ", "", 1)) _expect_invalid(ask.replace("主动回传。_", "随便处理。_", 1)) _expect_invalid(ask.replace("id/ctm-b91f2c9d4e71", "id/not-valid", 1)) _expect_invalid(ask.replace("type/ask", "type/ask/type/ask", 1)) _expect_invalid(ask.replace("return/push", "return/push/authority/granted", 1)) _expect_invalid(reply.replace("outcome/completed", "status/completed", 1)) _expect_invalid(ask.replace("[codex-task-messenger]: codex-task-messenger:v2/", "