432 lines
16 KiB
TypeScript
432 lines
16 KiB
TypeScript
import fs from "fs"
|
|
import os from "os"
|
|
import path from "path"
|
|
import { MatrixBotClient } from "./matrix-client.js"
|
|
import { SessionManager } from "./session-manager.js"
|
|
import { loadConfig } from "./config-loader.js"
|
|
import { setAppLog, log, logError } from "./logger.js"
|
|
|
|
// =============================================================================
|
|
// Plugin data directory — uses the same default as matrix-client
|
|
// =============================================================================
|
|
|
|
function ensureDataDir(dir: string): void {
|
|
if (!fs.existsSync(dir)) {
|
|
fs.mkdirSync(dir, { recursive: true })
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Global mutex — prevents double startup (server + client both load plugins)
|
|
// =============================================================================
|
|
|
|
let LOCK_FILE = ""
|
|
|
|
function tryAcquireLock(): boolean {
|
|
try {
|
|
// Check if an existing lock is stale (process no longer running)
|
|
if (fs.existsSync(LOCK_FILE)) {
|
|
const existingPid = parseInt(fs.readFileSync(LOCK_FILE, "utf-8").trim(), 10)
|
|
if (existingPid && existingPid !== process.pid) {
|
|
try {
|
|
process.kill(existingPid, 0) // throws if process doesn't exist
|
|
} catch {
|
|
// Process dead, remove stale lock
|
|
fs.unlinkSync(LOCK_FILE)
|
|
}
|
|
} else if (existingPid === process.pid) {
|
|
// Same process re-entering (shouldn't happen, but be safe)
|
|
return false
|
|
} else {
|
|
return false // Valid lock held by another process
|
|
}
|
|
}
|
|
fs.writeFileSync(LOCK_FILE, String(process.pid), { mode: 0o644 })
|
|
return true
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
function releaseLock(): void {
|
|
try {
|
|
fs.unlinkSync(LOCK_FILE)
|
|
} catch {
|
|
// ignore
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Plugin Setup
|
|
// =============================================================================
|
|
|
|
async function runPlugin(ctx: any, options: any, dataDir: string) {
|
|
ensureDataDir(dataDir)
|
|
LOCK_FILE = `${dataDir}/lock`
|
|
|
|
const directory = ctx?.location?.directory || process.cwd()
|
|
log(`Plugin setup started, directory=${directory}`)
|
|
|
|
// Acquire global lock — only one instance (server or client) starts the bot
|
|
const locked = tryAcquireLock()
|
|
if (!locked) {
|
|
log("Another instance already running (lock held), skipping bot startup")
|
|
return () => {}
|
|
}
|
|
|
|
const eventController = new AbortController()
|
|
let cleanup: (() => void) | undefined
|
|
|
|
try {
|
|
log(`Configuration loaded`)
|
|
|
|
if (!options.homeserver) {
|
|
log("Plugin disabled (no homeserver configured)")
|
|
return () => {}
|
|
}
|
|
|
|
log(`Configuration loaded`)
|
|
|
|
const matrix = new MatrixBotClient(options)
|
|
const sessionManager = new SessionManager(options.rateLimitSeconds || 5)
|
|
|
|
// Track Matrix context per OpenCode session
|
|
const sessionContextMap = new Map<string, { roomId: string; threadRootId: string; context: any }>()
|
|
|
|
// Pending prompt responses — maps sessionID -> { resolve, reject }
|
|
const pendingResponses = new Map<string, {
|
|
resolve: (text: string) => void
|
|
reject: (err: any) => void
|
|
}>()
|
|
|
|
// V2 Event subscription — logs all events for debugging
|
|
void (async () => {
|
|
try {
|
|
for await (const event of ctx.event.subscribe({ signal: eventController.signal })) {
|
|
// Log event type for debugging
|
|
let eventType = ""
|
|
let eventData: any = event
|
|
if (typeof event === "string") {
|
|
try {
|
|
eventData = JSON.parse(event)
|
|
eventType = eventData?.type || ""
|
|
} catch {
|
|
eventType = "(raw string)"
|
|
}
|
|
} else {
|
|
eventType = eventData?.type || ""
|
|
}
|
|
|
|
// Check for session completion events
|
|
const sessionId = eventData?.sessionID || eventData?.session?.id || ""
|
|
if (sessionId && pendingResponses.has(sessionId)) {
|
|
// Check if event indicates session completion
|
|
const outcome = eventData?.outcome || eventData?.data?.outcome || eventData?.update?.outcome
|
|
if (outcome && ["succeeded", "failed", "interrupted"].includes(outcome)) {
|
|
log(`Session ${sessionId} completed with outcome: ${outcome} (from event)`)
|
|
// Will be picked up by polling loop
|
|
}
|
|
}
|
|
|
|
// Log first 200 chars of interesting events
|
|
if (eventType.includes("session") && sessionId) {
|
|
const preview = typeof event === "string" ? event.slice(0, 200) : JSON.stringify(eventData).slice(0, 200)
|
|
log(`Event: type=${eventType}, session=${sessionId}, preview=${preview}`)
|
|
}
|
|
}
|
|
} catch (err) {
|
|
if (String(err).includes("AbortError") || (err as { name?: string }).name === "AbortError") {
|
|
// expected during cleanup
|
|
} else {
|
|
log(`Event subscription error: ${String(err)}`)
|
|
}
|
|
}
|
|
})()
|
|
|
|
// Message handler
|
|
matrix.on(async (data: any) => {
|
|
const { context, query, sender, roomId, eventId, timestamp } = data
|
|
|
|
// Check rate limit
|
|
if (sessionManager.isRateLimited(sender)) {
|
|
await matrix.sendNotice(context, "Rate limited. Please wait a moment.")
|
|
return
|
|
}
|
|
|
|
// Check for active query
|
|
if (sessionManager.hasActiveQuery(roomId, context.replyThreadRootId || context.eventId)) {
|
|
await matrix.sendNotice(context, "A request is already running in this thread. Please wait.")
|
|
return
|
|
}
|
|
|
|
const abortFn = () => { /* placeholder for cancel */ }
|
|
const releaseQuery = sessionManager.markQueryActive(roomId, context.replyThreadRootId || context.eventId, abortFn)
|
|
|
|
const threadRootId = context.replyThreadRootId || context.eventId
|
|
const session = sessionManager.getOrCreateSession(roomId, threadRootId, eventId)
|
|
session.isActive = true
|
|
session.messageCount++
|
|
session.inputChars += query.length
|
|
session.lastEventIds.set(context.replyThreadRootId || context.eventId, eventId)
|
|
session.lastActivity = Date.now()
|
|
|
|
let opencodeSessionId = session.opencodeSessionId
|
|
|
|
// If we have a session ID from the map, verify it exists in OpenCode
|
|
if (opencodeSessionId) {
|
|
try {
|
|
await ctx.session.get({ sessionID: opencodeSessionId })
|
|
log(`Using existing OpenCode session ${opencodeSessionId}`)
|
|
} catch (getErr: any) {
|
|
log(`Session ${opencodeSessionId} not found in OpenCode, creating new one`)
|
|
opencodeSessionId = ""
|
|
}
|
|
}
|
|
|
|
// If no session ID yet, create one
|
|
if (!opencodeSessionId) {
|
|
log(`Creating new OpenCode session for ${roomId}:${threadRootId}`)
|
|
try {
|
|
const createResult = await ctx.session.create({ title: `Matrix: ${roomId.slice(0, 30)}...` })
|
|
opencodeSessionId = createResult?.id || createResult?.data?.id
|
|
log(`Created OpenCode session ${opencodeSessionId} for ${roomId}:${threadRootId}`)
|
|
session.opencodeSessionId = opencodeSessionId
|
|
sessionManager.saveSession(roomId, threadRootId, opencodeSessionId)
|
|
sessionContextMap.set(opencodeSessionId, { roomId, threadRootId, context })
|
|
} catch (createErr: any) {
|
|
log(`Failed to create session: ${String(createErr)}`)
|
|
session.isActive = false
|
|
session.lastActivity = Date.now()
|
|
releaseQuery()
|
|
await matrix.sendNotice(context, "Failed to create session. Check logs.")
|
|
return
|
|
}
|
|
}
|
|
|
|
try {
|
|
// Trigger prompt — returns admission receipt, not response text
|
|
await ctx.session.prompt({
|
|
sessionID: opencodeSessionId,
|
|
text: query,
|
|
})
|
|
log(`Prompt sent to session ${opencodeSessionId}`)
|
|
|
|
// Poll session.get() to wait for completion (outcome field)
|
|
let responseText = ""
|
|
let lastAssistantText = ""
|
|
|
|
while (true) {
|
|
await new Promise((r) => setTimeout(r, 500))
|
|
|
|
try {
|
|
const sessionInfo = await ctx.session.get({ sessionID: opencodeSessionId })
|
|
const outcome = sessionInfo?.outcome || sessionInfo?.data?.outcome
|
|
|
|
// Session completed — collect response text from messages
|
|
if (outcome && ["succeeded", "failed", "interrupted"].includes(outcome)) {
|
|
log(`Session ${opencodeSessionId} completed with outcome: ${outcome}`)
|
|
|
|
// Collect assistant text from session messages
|
|
const messages = sessionInfo?.messages || sessionInfo?.data?.messages
|
|
if (Array.isArray(messages)) {
|
|
for (const msg of messages) {
|
|
if (msg?.type === "assistant" || msg?.role === "assistant") {
|
|
const content = msg?.content || msg?.text
|
|
if (typeof content === "string") {
|
|
lastAssistantText += content
|
|
} else if (Array.isArray(content)) {
|
|
for (const part of content) {
|
|
if (part?.type === "text" && part?.text) {
|
|
lastAssistantText += part.text
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Also check session metadata for response text
|
|
if (!lastAssistantText) {
|
|
const metaText = sessionInfo?.metadata?.responseText || sessionInfo?.data?.metadata?.responseText
|
|
if (metaText) lastAssistantText = String(metaText)
|
|
}
|
|
|
|
responseText = lastAssistantText || "No response text collected"
|
|
break
|
|
}
|
|
|
|
// Track assistant text as it streams (for partial responses)
|
|
const streamMessages = sessionInfo?.streamMessages || sessionInfo?.data?.streamMessages
|
|
if (Array.isArray(streamMessages)) {
|
|
for (const msg of streamMessages) {
|
|
if (msg?.type === "assistant" || msg?.role === "assistant") {
|
|
const content = msg?.content || msg?.text
|
|
if (typeof content === "string" && content !== lastAssistantText) {
|
|
lastAssistantText = content
|
|
} else if (Array.isArray(content)) {
|
|
let text = ""
|
|
for (const part of content) {
|
|
if (part?.type === "text" && part?.text) text += part.text
|
|
}
|
|
if (text && text !== lastAssistantText) lastAssistantText = text
|
|
}
|
|
}
|
|
}
|
|
}
|
|
} catch (getErr: any) {
|
|
log(`Session get error (will retry): ${String(getErr)}`)
|
|
// Session might be temporarily unavailable, keep polling
|
|
}
|
|
}
|
|
|
|
session.outputChars += responseText.length
|
|
await matrix.sendReply(context, responseText)
|
|
} catch (err: any) {
|
|
log(`Error processing message: ${String(err)}`)
|
|
const errMsg = err?.response?.error?.message || String(err)
|
|
await matrix.sendNotice(context, `Error: ${errMsg.slice(0, 200)}`)
|
|
} finally {
|
|
session.isActive = false
|
|
session.lastActivity = Date.now()
|
|
sessionContextMap.delete(opencodeSessionId)
|
|
releaseQuery()
|
|
}
|
|
})
|
|
|
|
// Bridge commands
|
|
async function handleBridgeCommand(
|
|
context: { roomId: string; replyThreadRootId?: string; threadRootId?: string; eventId: string },
|
|
cmdName: string,
|
|
sendFn: (text: string) => Promise<void>,
|
|
): Promise<void> {
|
|
const threadRootId = context.threadRootId || context.replyThreadRootId || context.eventId
|
|
const key = sessionManager.getSessionKey(context.roomId, threadRootId)
|
|
const sess = sessionManager.sessions.get(key)
|
|
|
|
switch (cmdName) {
|
|
case "status": {
|
|
if (!sess) {
|
|
await sendFn("No active session.")
|
|
return
|
|
}
|
|
const age = Math.round((Date.now() - sess.lastActivity) / 60000)
|
|
const msg = `Session status:\n- Messages: ${sess.messageCount}\n- Age: ~${age} min ago\n- Input: ~${sess.inputChars} chars\n- Output: ~${sess.outputChars} chars`
|
|
await sendFn(msg)
|
|
break
|
|
}
|
|
case "clear":
|
|
case "reset": {
|
|
sessionManager.removeSession(context.roomId, threadRootId)
|
|
await sendFn("Session cleared. Next message will start a fresh session.")
|
|
break
|
|
}
|
|
case "help":
|
|
case "h": {
|
|
const trigger = options.triggerPatterns?.[0] || "!oc "
|
|
const botName = options.botName || "opencode"
|
|
const msg = [
|
|
`${botName} - OpenCode Matrix Plugin`,
|
|
"",
|
|
"Bridge commands:",
|
|
`- /h or /help - Show this help`,
|
|
`- /status - Show current chat session info`,
|
|
`- /clear or /reset - Delete current session history`,
|
|
"",
|
|
`Usage: ${trigger} <your question>`,
|
|
`Or @${botName} <your question>`,
|
|
`Or reply in a thread`,
|
|
].join("\n")
|
|
await sendFn(msg)
|
|
break
|
|
}
|
|
default:
|
|
await sendFn(`Unknown command: ${cmdName}. Try /help`)
|
|
}
|
|
}
|
|
|
|
// Override the matrix.on handler to also process commands
|
|
const originalListener = (matrix as any)["listeners"][0]
|
|
;(matrix as any)["listeners"] = []
|
|
matrix.on(async (data: any) => {
|
|
const { context, query, sender, eventId } = data
|
|
|
|
// Handle bridge commands
|
|
if (query.startsWith("/")) {
|
|
const cmdName = query.slice(1).split(" ")[0].toLowerCase()
|
|
const bridgeCommands = ["status", "clear", "reset", "help", "h"]
|
|
if (bridgeCommands.includes(cmdName)) {
|
|
const sendFn = (text: string) => matrix.sendNotice(context, text)
|
|
await handleBridgeCommand(context, cmdName, sendFn)
|
|
return
|
|
}
|
|
}
|
|
|
|
// Forward to original handler
|
|
if (originalListener) {
|
|
await originalListener(data)
|
|
}
|
|
})
|
|
|
|
// V2 Storage
|
|
if (ctx?.storage) {
|
|
await ctx.storage.set("matrix/plugin_version", "1.0.0")
|
|
await ctx.storage.set("matrix/started_at", new Date().toISOString())
|
|
}
|
|
|
|
// Start Matrix client
|
|
await matrix.start()
|
|
|
|
log(`Plugin setup completed successfully`)
|
|
|
|
cleanup = () => {
|
|
log("Plugin cleaning up...")
|
|
eventController.abort()
|
|
matrix.stop()
|
|
// Don't release lock — keep it on disk so client always sees it
|
|
}
|
|
|
|
return cleanup
|
|
} catch (err) {
|
|
logError(`Setup error: ${String(err)}`)
|
|
releaseLock()
|
|
return () => {}
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// V2 Setup Function
|
|
// =============================================================================
|
|
|
|
async function v2Setup(ctx: any) {
|
|
const rawStorage = (ctx.options as any)?.storagePath
|
|
const dataDir = rawStorage
|
|
? path.join(rawStorage, "plugin-data")
|
|
: path.join(os.homedir(), ".local", "share", "opencode-matrix-bot", "plugin-data")
|
|
ensureDataDir(dataDir)
|
|
|
|
// Try to use OpenCode's logging API
|
|
if (ctx?.app?.log) {
|
|
setAppLog(ctx.app.log.bind(ctx.app))
|
|
}
|
|
|
|
try {
|
|
const { options } = await loadConfig(ctx.options)
|
|
return await runPlugin(ctx, options, dataDir)
|
|
} catch (err) {
|
|
logError(`V2 setup error: ${String(err)}`)
|
|
return () => {}
|
|
}
|
|
}
|
|
|
|
// =============================================================================
|
|
// Export
|
|
// =============================================================================
|
|
|
|
export default {
|
|
id: "matrix-plugin",
|
|
async setup(ctx: any) {
|
|
log("Using V2 path (full context)")
|
|
return await v2Setup(ctx)
|
|
},
|
|
}
|