"""Run as agent inside a disposable Slack Harbor container, before any policy.

write: discover four tools, read a thread, write one reply, read it back.
after-reset: open a fresh MCP session and prove the original thread is restored.
Uses the MCP SDK already installed in the pinned image. Never retries a write.
"""

import argparse
import asyncio
import csv
import io
import json
import os
from pathlib import Path
import pwd
import stat
import uuid

PUBLIC = Path("/run/twinenv/mcp.json")


def require(condition, message):
    if not condition:
        raise RuntimeError(message)


def read_manifest():
    agent = pwd.getpwnam("agent").pw_uid
    owner = pwd.getpwnam("twinenv").pw_uid
    require(os.geteuid() == agent and agent not in (0, owner), "run as agent")
    for path in [*reversed(PUBLIC.parents), PUBLIC]:
        info = path.lstat()
        require(not stat.S_ISLNK(info.st_mode), "manifest path contains a symlink")
        require(info.st_uid in (0, owner) and not info.st_mode & 0o022
                and not os.access(path, os.W_OK), "manifest path is agent-writable")
        require(stat.S_ISREG(info.st_mode) if path == PUBLIC else stat.S_ISDIR(info.st_mode),
                "invalid manifest path")
    with PUBLIC.open("rb") as source:
        raw = source.read(16385)
    require(len(raw) <= 16384, "manifest exceeds size limit")
    return raw


def table(result):
    require(not result.isError and len(result.content) == 1, "MCP tool failed")
    block = result.content[0]
    require(block.type == "text", "expected text content")
    rows = list(csv.reader(io.StringIO(block.text)))
    require(rows and all(len(row) == len(rows[0]) for row in rows), "invalid CSV result")
    return [dict(zip(rows[0], row)) for row in rows[1:]]


async def exercise(session, pins, report, phase, receipt):
    await session.initialize()
    listing = await session.list_tools()
    require(sorted(t.name for t in listing.tools) == sorted(pins["tools"]),
            "unexpected tool list")

    async def call(name, arguments):
        result = await session.call_tool(name, arguments, read_timeout_seconds=120.0)
        require(not result.isError, "MCP tool failed; reset before trying again")
        return result

    if phase == "write":
        require(not receipt.exists(), "receipt already exists; use a fresh container")
        channels = table(await call("channels_list", {"channel_types": "public_channel", "limit": 100}))
        channel = next(row["ID"] for row in channels if row["Name"] == "#refund-desk")
        history = table(await call("conversations_history", {"channel_id": channel, "limit": "100"}))
        parent = next(row["MsgID"] for row in history if row.get("ThreadTs") == row["MsgID"])
        args = {"channel_id": channel, "thread_ts": parent, "limit": "100"}
        before = table(await call("conversations_replies", args))
        marker = "Rystic Harbor setup check " + uuid.uuid4().hex
        await call("conversations_add_message", {
            "channel_id": channel, "thread_ts": parent,
            "text": marker, "content_type": "text/plain",
        })
        after = table(await call("conversations_replies", args))
        require(len(after) == len(before) + 1
                and sum(marker in row.values() for row in after) == 1,
                "reply was not read back exactly once")
        receipt.write_text(json.dumps({"generation": report["generation"], "args": args,
                                      "before": before, "marker": marker}))
    else:
        saved = json.loads(receipt.read_text())
        require(report["generation"] != saved["generation"], "episode generation did not change")
        after = table(await call("conversations_replies", saved["args"]))
        require(after == saved["before"], "reset did not restore the original thread")
    return {"ok": True, "phase": phase, "generation": report["generation"],
            "tools": sorted(pins["tools"])}


async def main():
    # Imports stay here so pure checks can run without the image's dependencies.
    import httpx2
    from mcp import ClientSession
    from mcp.client.streamable_http import streamable_http_client
    from twinenv.mcp_harness import MCPHarnessConfig, validate_manifest

    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("phase", choices=("write", "after-reset"))
    parser.add_argument("--pins", type=Path, default=Path(__file__).with_name("pins.json"))
    parser.add_argument("--receipt", type=Path, default=Path("/app/mcp-smoke.json"))
    args = parser.parse_args()
    pins = json.loads(args.pins.read_text())
    config = MCPHarnessConfig(**{"expected_" + key: pins[key] for key in
                                ("profile_id", "profile_sha256", "tool_manifest_sha256")})
    raw = read_manifest()
    report = validate_manifest(raw, config)
    async with httpx2.AsyncClient(trust_env=False, timeout=120.0) as client:
        async with streamable_http_client(report["url"], http_client=client) as (read, write):
            async with ClientSession(read, write, read_timeout_seconds=120.0) as session:
                result = await exercise(session, pins, report, args.phase, args.receipt)
                require(read_manifest() == raw, "episode changed during the check")
                print(json.dumps(result))


if __name__ == "__main__":
    asyncio.run(main())