Import a kimi-code session into opencode's SQLite storage (v1.18.x stable DB)

kimi_import.py

#!/usr/bin/env python3
"""Import a kimi-code session into opencode's SQLite storage (v1.18.x stable DB).

Converts ~/.kimi-code/sessions/<wd>/<session_id>/{state.json,agents/main/wire.jsonl}
into opencode message/part rows so the session appears in opencode and can be forked.
"""
import json
import random
import sqlite3
import sys
from pathlib import Path

KIMI_SESSION = Path(sys.argv[1])
OPENCODE_DB = Path(sys.argv[2])
PROJECT_ID = sys.argv[3]
OPENCODE_VERSION = "1.18.18"

B62 = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz"

id_state = {"ts": 0, "counter": 0}

def ascending_id(prefix: str, ts: int) -> str:
    if ts > id_state["ts"]:
        id_state["ts"] = ts
        id_state["counter"] = 0
    id_state["counter"] += 1
    value = id_state["ts"] * 0x1000 + id_state["counter"]
    hexpart = "".join(f"{(value >> (40 - 8 * i)) & 0xFF:02x}" for i in range(6))
    rand = "".join(random.choice(B62) for _ in range(14))
    return f"{prefix}_{hexpart}{rand}"

# ---------- load kimi session ----------
state = json.loads((KIMI_SESSION / "state.json").read_text())
wire_path = KIMI_SESSION / "agents" / "main" / "wire.jsonl"
events = [json.loads(l) for l in wire_path.read_text().splitlines() if l.strip()]

cwd = state["cwd"]
created = state["createdAt"]
updated = state["updatedAt"]
title = state.get("title") or f"Imported kimi-code session {state['id']}"

MODEL = {"providerID": "nebul", "modelID": "moonshotai/Kimi-K3"}
AGENT = "build"

# ---------- convert wire into opencode messages ----------
# opencode model: one user message per prompt; one assistant message per LLM step.
# assistant messages hold parts: step-start, reasoning/text/tool..., step-finish.
FINISH_MAP = {"tool_use": "tool-calls", "end_turn": "stop"}

messages = []  # list of dicts: {info, parts}
current = None  # open assistant message
pending_tools = {}  # uuid -> part ref


def close_step(ev=None):
    global current
    if current is None:
        return
    finish_reason = ev["event"].get("finishReason") if ev else None
    usage = (ev["event"].get("usage") if ev else None) or {}
    info = current["info"]
    info["time"]["completed"] = ev["time"] if ev else info["time"]["created"]
    tokens = {
        "input": usage.get("inputOther", 0),
        "output": usage.get("output", 0),
        "reasoning": 0,
        "cache": {
            "read": usage.get("inputCacheRead", 0),
            "write": usage.get("inputCacheCreation", 0),
        },
    }
    tokens["total"] = tokens["input"] + tokens["output"]
    info["tokens"] = tokens
    if finish_reason == "interrupted":
        info["error"] = {
            "name": "MessageAbortedError",
            "data": {"message": "The user interrupted the message"},
        }
    elif finish_reason is not None:
        info["finish"] = FINISH_MAP.get(finish_reason, "stop")
        current["parts"].append({
            "data": {
                "type": "step-finish",
                "reason": FINISH_MAP.get(finish_reason, "stop"),
                "cost": 0,
                "tokens": tokens,
            },
            "ts": ev["time"],
        })
    # parts get sequential ids/times in logical order
    for i, p in enumerate(current["parts"]):
        ts = max(p["ts"], current["info"]["time"]["created"]) + i
        pid = ascending_id("prt", ts)
        p["id"] = pid
        p["row_time"] = ts
        p["data"] = {**p["data"]}  # data excludes id/sessionID/messageID (they're columns)
    messages.append(current)
    current = None


last_user_id = None

for ev in events:
    t = ev.get("type")
    ts = ev.get("time", created)

    if t == "context.append_message":
        m = ev["message"]
        if m.get("role") != "user" or not m.get("id"):
            continue
        if (m.get("origin") or {}).get("kind") not in ("user", None):
            continue  # skip injections/task notifications
        close_step()
        text = "".join(c.get("text", "") for c in m.get("content", []) if c.get("type") == "text")
        if not text.strip():
            continue
        mid = ascending_id("msg", ts)
        last_user_id = mid
        messages.append({
            "info": {
                "role": "user",
                "time": {"created": ts},
                "agent": AGENT,
                "model": MODEL,
                "summary": {"diffs": []},
            },
            "parts": [{"data": {"type": "text", "text": text, "time": {"start": ts, "end": ts}}, "ts": ts}],
            "id": mid,
            "row_time": ts,
        })
        # user message parts: assign id now (single text part)
        messages[-1]["parts"][0]["id"] = ascending_id("prt", ts)
        messages[-1]["parts"][0]["row_time"] = ts

    elif t == "context.append_loop_event":
        e = ev["event"]
        et = e["type"]
        if et == "step.begin":
            close_step()
            if last_user_id is None:
                continue  # step before any prompt; skip defensively
            current = {
                "info": {
                    "parentID": last_user_id,
                    "role": "assistant",
                    "mode": AGENT,
                    "agent": AGENT,
                    "variant": "max",
                    "path": {"cwd": cwd, "root": cwd},
                    "cost": 0,
                    "tokens": {"input": 0, "output": 0, "reasoning": 0, "cache": {"read": 0, "write": 0}},
                    "modelID": MODEL["modelID"],
                    "providerID": MODEL["providerID"],
                    "time": {"created": ts},
                },
                "parts": [{"data": {"type": "step-start"}, "ts": ts}],
                "id": ascending_id("msg", ts),
                "row_time": ts,
                "uuid": e["uuid"],
            }
        elif et == "content.part" and current is not None:
            p = e["part"]
            if p.get("type") == "think":
                current["parts"].append({
                    "data": {"type": "reasoning", "text": p.get("think", ""), "time": {"start": ts, "end": ts}},
                    "ts": ts,
                })
            elif p.get("type") == "text":
                current["parts"].append({
                    "data": {"type": "text", "text": p.get("text", ""), "time": {"start": ts, "end": ts}},
                    "ts": ts,
                })
        elif et == "tool.call" and current is not None:
            part = {
                "data": {
                    "type": "tool",
                    "callID": e.get("toolCallId", e["uuid"]),
                    "tool": e.get("name", "unknown"),
                    "state": {
                        "status": "pending",
                        "input": e.get("args", {}),
                        "raw": json.dumps(e.get("args", {})),
                    },
                },
                "ts": ts,
                "call_ts": ts,
            }
            pending_tools[e["uuid"]] = part
            current["parts"].append(part)
        elif et == "tool.result":
            part = pending_tools.pop(e.get("parentUuid"), None)
            if part is None:
                continue
            res = e.get("result", {})
            output = res.get("output", "")
            start = part.pop("call_ts")
            if res.get("isError"):
                part["data"]["state"] = {
                    "status": "error",
                    "input": part["data"]["state"]["input"],
                    "error": output,
                    "time": {"start": start, "end": ts},
                }
            else:
                part["data"]["state"] = {
                    "status": "completed",
                    "input": part["data"]["state"]["input"],
                    "output": output,
                    "title": "",
                    "metadata": {},
                    "time": {"start": start, "end": ts},
                }
        elif et == "step.end":
            close_step(ev)

close_step()

# ---------- aggregate session tokens ----------
tot = {"input": 0, "output": 0, "reasoning": 0, "cache_read": 0, "cache_write": 0}
for m in messages:
    if m["info"]["role"] == "assistant":
        tk = m["info"].get("tokens", {})
        tot["input"] += tk.get("input", 0)
        tot["output"] += tk.get("output", 0)
        tot["reasoning"] += tk.get("reasoning", 0)
        tot["cache_read"] += tk.get("cache", {}).get("read", 0)
        tot["cache_write"] += tk.get("cache", {}).get("write", 0)

session_id = ascending_id("ses", created)
slug = "imported-kimi-" + "".join(random.choice(B62).lower() for _ in range(4))

session_info = {
    "id": session_id,
    "slug": slug,
    "projectID": PROJECT_ID,
    "directory": cwd,
    "path": "",
    "cost": 0,
    "tokens": {
        "input": tot["input"],
        "output": tot["output"],
        "reasoning": tot["reasoning"],
        "cache": {"read": tot["cache_read"], "write": tot["cache_write"]},
    },
    "title": title,
    "agent": AGENT,
    "model": {"id": MODEL["modelID"], "providerID": MODEL["providerID"], "variant": "max"},
    "version": OPENCODE_VERSION,
    "time": {"created": created, "updated": updated},
}

# ---------- write to opencode db ----------
db = sqlite3.connect(f"file:{OPENCODE_DB}?busy_timeout=10000", uri=True)
cur = db.cursor()
cur.execute("BEGIN IMMEDIATE")

cur.execute(
    """INSERT INTO session (id, project_id, workspace_id, parent_id, slug, directory, path,
       title, version, share_url, summary_additions, summary_deletions, summary_files,
       summary_diffs, metadata, cost, tokens_input, tokens_output, tokens_reasoning,
       tokens_cache_read, tokens_cache_write, revert, permission, agent, model,
       time_created, time_updated, time_compacting, time_archived)
       VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
    (
        session_id, PROJECT_ID, None, None, slug, cwd, "", title, OPENCODE_VERSION, None,
        None, None, None, None, None, 0.0, tot["input"], tot["output"], tot["reasoning"],
        tot["cache_read"], tot["cache_write"], None, None, AGENT,
        json.dumps({"id": MODEL["modelID"], "providerID": MODEL["providerID"], "variant": "max"}),
        created, updated, None, None,
    ),
)

seq = 0


def push_event(etype, data):
    global seq
    eid = ascending_id("evt", id_state["ts"])
    cur.execute(
        "INSERT INTO event (id, aggregate_id, seq, type, data) VALUES (?,?,?,?,?)",
        (eid, session_id, seq, etype, json.dumps(data, separators=(",", ":"))),
    )
    seq += 1


push_event("session.created.1", {"sessionID": session_id, "info": session_info})

for m in messages:
    mid = m["id"]
    info = m["info"]
    cur.execute(
        "INSERT INTO message (id, session_id, time_created, time_updated, data) VALUES (?,?,?,?,?)",
        (mid, session_id, m["row_time"], m["row_time"], json.dumps(info, separators=(",", ":"))),
    )
    push_event("message.updated.1", {
        "sessionID": session_id,
        "info": {**info, "id": mid, "sessionID": session_id},
    })
    for p in m["parts"]:
        cur.execute(
            "INSERT INTO part (id, message_id, session_id, time_created, time_updated, data) VALUES (?,?,?,?,?,?)",
            (p["id"], mid, session_id, p["row_time"], p["row_time"], json.dumps(p["data"], separators=(",", ":"))),
        )
        push_event("message.part.updated.1", {
            "sessionID": session_id,
            "part": {**p["data"], "id": p["id"], "sessionID": session_id, "messageID": mid},
            "time": p["row_time"],
        })

push_event("session.updated.1", {"sessionID": session_id, "info": session_info})

cur.execute(
    "INSERT INTO event_sequence (aggregate_id, seq, owner_id) VALUES (?,?,NULL)",
    (session_id, seq - 1),
)

db.commit()

# verify
n_msg = cur.execute("SELECT COUNT(*) FROM message WHERE session_id=?", (session_id,)).fetchone()[0]
n_part = cur.execute("SELECT COUNT(*) FROM part WHERE session_id=?", (session_id,)).fetchone()[0]
db.close()
print(f"imported session: {session_id}")
print(f"slug: {slug}")
print(f"messages: {n_msg}, parts: {n_part}, events: {seq}")