Plugin Conformance Testing

CopyCORE_ROOT = Path(__file__).resolve().parents[1]
REPOSITORY_ROOT = CORE_ROOT.parents[1]
PYTHON_HOSTS = ("claude-code", "cursor", "codex", "kimi", "antigravity")
GROUPS = (
    "python-bundles",
    "python-tests",
    "typescript-core",
    "openclaw",
    "opencode",
    "pi-agent",
    "deepseek",
)
LIVE_GROUP = "live-platform"
 
sys.path.insert(0, str(CORE_ROOT))
sys.path.insert(0, str(CORE_ROOT / "python"))
from memory_core import redact  # noqa: E402
from build.build import build  # noqa: E402
from conformance.artifacts import TYPESCRIPT_ARTIFACTS, verify_artifact as _typescript_artifact_check  # noqa: E402

Copydef _planned_checks(groups: set[str]) -> list[dict[str, Any]]:
    checks: list[dict[str, Any]] = []
    if "python-bundles" in groups:
        checks.append(
            {
                "name": "python-bundles",
                "group": "python-bundles",
                "status": "planned",
                "hosts": list(PYTHON_HOSTS),
                "kinds": {"native": list(PYTHON_HOSTS), "portable": ["mem0-agent-plugin"]},
            }
        )
    for group, commands in _runtime_commands().items():
        if group not in groups:
            continue
        for index, command in enumerate(commands, start=1):
            checks.append(
                {
                    "name": f"{group}-{index}",
                    "group": group,
                    "status": "planned",
                    "command": command,
                }
            )
        if group in TYPESCRIPT_ARTIFACTS:
            _, required = TYPESCRIPT_ARTIFACTS[group]
            checks.append(
                {
                    "name": f"{group}-artifact",
                    "group": group,
                    "status": "planned",
                    "required": list(required),
                }
            )
    return checks

Copydef _command_check(
    name: str,
    group: str,
    command: list[str],
    *,
    cwd: Path,
    environment_overrides: dict[str, str] | None = None,
) -> dict[str, Any]:
    started = time.monotonic()
    environment = dict(os.environ)
    environment.update(environment_overrides or {})
    safe_command = [redact(part) for part in command]
    try:
        result = subprocess.run(
            command,
            cwd=cwd,
            env=environment,
            text=True,
            capture_output=True,
            check=False,
        )
        output = redact((result.stdout + result.stderr).strip())
        return {
            "name": name,
            "group": group,
            "status": "passed" if result.returncode == 0 else "failed",
            "duration_seconds": round(time.monotonic() - started, 3),
            "command": safe_command,
            "exit_code": result.returncode,
            **({"output": output[-8_000:]} if output else {}),
        }
    except OSError as exc:
        return {
            "name": name,
            "group": group,
            "status": "failed",
            "duration_seconds": round(time.monotonic() - started, 3),
            "command": safe_command,
            "exit_code": None,
            "output": redact(str(exc)),
        }

CopyTYPESCRIPT_ARTIFACTS = {
    "openclaw": (INTEGRATIONS_ROOT / "openclaw", ("dist/index.js", "dist/index.d.ts")),
    "opencode": (INTEGRATIONS_ROOT / "opencode-plugin", ("dist/index.js", "index.d.ts")),
    "pi-agent": (
        INTEGRATIONS_ROOT / "pi-agent-plugin",
        ("dist/index.js", "dist/index.d.ts", "dist/entry.js", "dist/entry.d.ts"),
    ),
    "deepseek": (INTEGRATIONS_ROOT / "deepseek-plugin", ("dist/index.js", "dist/index.d.ts")),
}

CopyMONOREPO_IMPORT = re.compile(
    r"(?:from\s+|import\s*\(|require\s*\()\s*['\"][^'\"]*agent-plugin-core"
)
 
 
def verify_artifact(group: str, package: Path, required: tuple[str, ...]) -> dict[str, Any]:
    started = time.monotonic()
    errors = [f"missing package artifact: {name}" for name in required if not (package / name).is_file()]
    dist = package / "dist"
    for pattern in ("*.js", "*.mjs", "*.cjs", "*.d.ts"):
        for artifact in dist.rglob(pattern) if dist.is_dir() else ():
            if MONOREPO_IMPORT.search(artifact.read_text(encoding="utf-8")):
                errors.append(f"monorepo source import in package artifact: {artifact.relative_to(package)}")
    return {
        "name": f"{group}-artifact",
        "group": group,
        "status": "failed" if errors else "passed",
        "duration_seconds": round(time.monotonic() - started, 3),
        **({"output": "\n".join(errors)} if errors else {}),
    }

Copydef read_hook_input() -> dict:
    try:
        value = json.load(sys.stdin)
        return value if isinstance(value, dict) else {}
    except (json.JSONDecodeError, OSError):
        return {}

CopySTALE_RUNNING_SECONDS = 300
PENDING_EXPIRY_SECONDS = 7 * 24 * 60 * 60
PENDING_LAUNCH_LIMIT = 5
DEFAULT_IDLE_FLUSH_SECONDS = 300

Copydef first_prompt_memory_output(store: EvidenceStore, hook_input: dict) -> dict:
    """Search once before the agent handles the first prompt in a session."""
    repo, session_id, prompt, is_first_prompt = record_user_prompt(store, hook_input)
    if not is_first_prompt:
        return {}
    try:
        minimum_query_chars = int(os.environ.get("MEM0_CODE_MIN_QUERY_CHARS", "20"))
    except ValueError:
        minimum_query_chars = 20
    if len(prompt.strip()) < max(minimum_query_chars, 1):
        return {}
    result = search_memories(
        store, repo, session_id, bounded(prompt, 6000),
        top_k=5, operation="first-prompt-search", timeout=2,
    )

Copydef _read_transcript(path: str) -> list[dict[str, str]]:
    messages = []
    try:
        with open(path, encoding="utf-8") as transcript:
            for line in transcript:
                try:
                    step = json.loads(line)
                except json.JSONDecodeError:
                    continue
                content = step.get("content")
                if not isinstance(content, str) or step.get("status") != "DONE":
                    continue
                if step.get("type") == "USER_INPUT":
                    match = re.search(r"<USER_REQUEST>\s*(.*?)\s*</USER_REQUEST>", content, re.DOTALL)
                    messages.append({"role": "user", "content": redact(match.group(1) if match else content).strip()})
                elif step.get("source") == "MODEL" and step.get("type") == "PLANNER_RESPONSE":
                    messages.append({"role": "assistant", "content": redact(content).strip()})
    except (OSError, TypeError):
        pass
    return messages

Copydef _session_dir(session_id: str) -> Path | None:
    home = Path(os.environ.get("KIMI_CODE_HOME", Path.home() / ".kimi-code")).expanduser()
    found = None
    try:
        with (home / "session_index.jsonl").open(encoding="utf-8") as index:
            for line in index:
                try:
                    entry = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if entry.get("sessionId") == session_id and isinstance(entry.get("sessionDir"), str):
                    found = Path(entry["sessionDir"])
    except OSError:
        pass
    if found:
        return found
    if not session_id or Path(session_id).name != session_id:
        return None
    return next((path for path in (home / "sessions").glob(f"*/{session_id}") if path.is_dir()), None)
 
 
def _last_assistant_message(transcript: Path) -> str:
    message = ""
    step_id = None
    parts = []
    try:
        lines = transcript.read_text(encoding="utf-8").splitlines()
    except OSError:
        return ""
    for line in lines:
        try:
            record = json.loads(line)
        except json.JSONDecodeError:
            continue
        if record.get("type") != "context.append_loop_event" or not isinstance(record.get("event"), dict):
            continue
        event = record["event"]
        if event.get("type") == "step.begin":
            step_id = event.get("uuid")
            parts = []
        elif event.get("type") == "content.part" and event.get("stepUuid") == step_id:
            part = event.get("part")
            if isinstance(part, dict) and part.get("type") == "text" and isinstance(part.get("text"), str):
                parts.append(part["text"])
        elif event.get("type") == "step.end" and event.get("uuid") == step_id:
            if event.get("finishReason") not in {"error", "interrupted"} and parts:
                message = "".join(parts)
            step_id = None
            parts = []
    return message

Copydef main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--plugin-data-dir", default="")
    parser.add_argument("--harness", default="")
    subparsers = parser.add_subparsers(dest="command", required=True)
 
    status = subparsers.add_parser("status")
    status.add_argument("--json", action="store_true")
 
    doctor_parser = subparsers.add_parser("doctor")
    doctor_parser.add_argument("--json", action="store_true")
 
    subparsers.add_parser("pause")
    subparsers.add_parser("resume")
 
    forget = subparsers.add_parser("forget")
    forget.add_argument("--remote", action="store_true")
    forget.add_argument("--yes", action="store_true")
    forget.add_argument("--include-project-memory", action="store_true")
 
    args = parser.parse_args()
    if args.harness:
        source_tag = f"{args.harness.replace('-', '_')}_plugin"
        configure_harness(args.harness, source_tag=source_tag)
        telemetry.init(harness=args.harness, source_tag=source_tag.upper())
    if args.plugin_data_dir: