Compare commits

..

6 Commits

12 changed files with 1058 additions and 129 deletions

View File

@@ -39,4 +39,13 @@ Use Bun exclusively. Do not use npm or yarn.
- Branch: `feat/<task>`, `fix/<task>`, `ci/<task>`, etc.
- Commits: conventional commit format, title <= 50 chars
- Signing: If `GPG_PRIVATE_KEY_PASSPHRASE` env var is available, use it to sign commits with `git commit -S`
## Nix
Use the Nix dev shell for project commands by default.
- Run repo tooling through `nix develop -c`, e.g. `nix develop -c bun test`.
- Use Bun exclusively inside the Nix shell.
- Do not use host `bun`, `node`, `tsc`, or package binaries for project tasks unless explicitly checking host behavior.
- Simple inspection commands like `rg`, `sed`, `ls`, and `git status` may run outside Nix.
- While `flake.nix` is untracked, use `nix develop path:. -c <command>`.

View File

@@ -7,5 +7,9 @@
"format": "oxfmt --write .",
"start": "bun run src/agent-test-cli.ts",
"typecheck": "bun tsc --noEmit"
},
"dependencies": {
"@freya/agent-protocol": "workspace:*",
"@nym.sh/jrpc": "^0.1.0"
}
}

View File

@@ -1,3 +1,13 @@
import type {
AgentClientApi,
AgentEvent,
AgentServerApi,
SendMessageResult,
} from "@freya/agent-protocol"
import type { JrpcChannel, JrpcMessage, JsonRpcMessage } from "@nym.sh/jrpc"
import { JsonRpcClient, JsonRpcServer } from "@nym.sh/jrpc"
type JsonObject = Record<string, unknown>
interface AuthUser {
@@ -15,10 +25,6 @@ interface AuthSession {
}
}
interface QueryResponse {
message: string
}
interface QueryToolDefinition {
name: string
label: string
@@ -60,6 +66,219 @@ class CookieJar {
}
}
class AgentWebSocketSession implements AgentClientApi {
private readonly channel: WebSocketJrpcChannel
private readonly client: JsonRpcClient<AgentServerApi>
private readonly server: JsonRpcServer<AgentClientApi>
private conversationId: string | undefined
private responseHadText = false
private constructor(channel: WebSocketJrpcChannel) {
this.channel = channel
this.client = new JsonRpcClient<AgentServerApi>(channel)
this.server = new JsonRpcServer<AgentClientApi>(
{
notify: this.notify.bind(this),
},
channel,
)
}
static async connect(backendUrl: string, cookies: CookieJar): Promise<AgentWebSocketSession> {
const channel = new WebSocketJrpcChannel(agentWebSocketUrl(backendUrl), cookies.header())
const session = new AgentWebSocketSession(channel)
try {
await channel.waitUntilOpen()
void session.server.start().catch((err: unknown) => {
if (!channel.isClosed()) {
console.error(`\nWebSocket JSON-RPC failed: ${formatError(err)}\n`)
}
})
await session.client.call("ping")
} catch (err) {
channel.close()
throw err
}
return session
}
async ask(message: string): Promise<void> {
this.responseHadText = false
const result = await this.sendMessage(message)
if (result.conversationId) {
this.conversationId = result.conversationId
}
if (!this.responseHadText) {
console.log(`\nagent> ${result.message || "(no message)"}`)
}
console.log("")
}
notify(event: AgentEvent): void {
switch (event.type) {
case "conversation_started":
this.conversationId = event.conversationId
break
case "message_created":
this.printMessage(event.text)
break
case "tool_started":
console.log(`\ntool> ${event.toolName} started`)
break
case "tool_finished":
console.log(`tool> ${event.toolName} ${event.ok ? "finished" : "failed"}`)
break
case "message_finished":
break
case "message_failed":
console.log(`\nagent! ${event.error}`)
break
}
}
describeConversation(): string {
return this.conversationId ? `Conversation: ${this.conversationId}` : "No conversation yet."
}
close(): void {
this.channel.close()
}
private async sendMessage(message: string): Promise<SendMessageResult> {
return this.client.call("sendMessage", message)
}
private printMessage(text: string): void {
if (text === "") return
console.log(`\nagent> ${text}`)
this.responseHadText = true
}
}
class WebSocketJrpcChannel implements JrpcChannel {
private readonly ws: WebSocket
private readonly opened: Promise<void>
private closed = false
private openedOnce = false
private queue: JrpcMessage[] = []
private waiters: Array<(result: IteratorResult<JrpcMessage, void>) => void> = []
constructor(url: string, cookieHeader?: string) {
this.ws = new WebSocket(url, createWebSocketOptions(cookieHeader))
this.opened = new Promise((resolve, reject) => {
this.ws.onopen = () => {
this.openedOnce = true
resolve()
}
this.ws.onerror = () => {
if (!this.openedOnce) {
reject(new Error(`Could not connect to ${url}`))
}
}
this.ws.onclose = (event) => {
if (!this.openedOnce) {
reject(new Error(formatWebSocketClose(url, event)))
}
this.close()
}
})
this.ws.onmessage = (event) => {
this.receive(event.data)
}
}
waitUntilOpen(): Promise<void> {
return this.opened
}
isClosed(): boolean {
return this.closed
}
async send(msg: JsonRpcMessage): Promise<void> {
await this.opened
if (this.closed || this.ws.readyState !== WebSocket.OPEN) {
throw new Error("JSON-RPC WebSocket channel is closed")
}
this.ws.send(JSON.stringify(msg))
}
async next(): Promise<IteratorResult<JrpcMessage, void>> {
const msg = this.queue.shift()
if (msg) {
return { done: false, value: msg }
}
if (this.closed) {
return { done: true, value: undefined }
}
return new Promise((resolve) => {
this.waiters.push(resolve)
})
}
async return(): Promise<IteratorResult<JrpcMessage, void>> {
this.close()
return { done: true, value: undefined }
}
async throw(error?: unknown): Promise<IteratorResult<JrpcMessage, void>> {
this.close()
throw error
}
async [Symbol.asyncDispose](): Promise<void> {
await this.return()
}
close(): void {
if (this.closed) return
this.closed = true
for (const resolve of this.waiters.splice(0)) {
resolve({ done: true, value: undefined })
}
if (this.ws.readyState === WebSocket.CONNECTING || this.ws.readyState === WebSocket.OPEN) {
this.ws.close()
}
}
[Symbol.asyncIterator](): AsyncGenerator<JrpcMessage, void, unknown> {
return this
}
private receive(message: unknown): void {
const parsed = parseJrpcMessage(message)
if (!parsed) {
this.ws.close(1003, "Invalid JSON-RPC message")
this.close()
return
}
this.push(parsed)
}
private push(msg: JrpcMessage): void {
if (this.closed) return
const resolve = this.waiters.shift()
if (resolve) {
resolve({ done: false, value: msg })
return
}
this.queue.push(msg)
}
}
async function main(): Promise<void> {
if (wantsHelp()) {
printUsage()
@@ -111,8 +330,11 @@ async function runChatLoop(
cookies: CookieJar,
session: AuthSession,
): Promise<void> {
const agent = await AgentWebSocketSession.connect(backendUrl, cookies)
console.log("Connected to /api/agent/ws")
printHelp()
try {
for (;;) {
const input = askOptional("you> ")?.trim()
if (!input) continue
@@ -132,13 +354,20 @@ async function runChatLoop(
continue
}
if (input === "/conversation") {
console.log(agent.describeConversation())
continue
}
if (input === "/tools") {
await runCliCommand(() => listQueryTools(backendUrl, cookies))
continue
}
if (input.startsWith("/tool ")) {
await runCliCommand(() => executeQueryTool(backendUrl, cookies, input.slice("/tool ".length)))
await runCliCommand(() =>
executeQueryTool(backendUrl, cookies, input.slice("/tool ".length)),
)
continue
}
@@ -157,25 +386,14 @@ async function runChatLoop(
}
try {
await askAgent(backendUrl, cookies, input)
await agent.ask(input)
} catch (err) {
console.error(`\n${formatError(err)}\n`)
}
}
}
async function askAgent(backendUrl: string, cookies: CookieJar, message: string): Promise<void> {
const data = await requestJson(backendUrl, cookies, "/api/agent", {
method: "POST",
body: { message },
})
if (!isQueryResponse(data)) {
throw new Error("Query returned an unexpected response shape")
} finally {
agent.close()
}
console.log(`\nagent> ${data.message || "(no message)"}`)
console.log("")
}
async function runCliCommand(command: () => Promise<void>): Promise<void> {
@@ -327,7 +545,7 @@ async function requestJson(
function printIntro(): void {
console.log("FREYA agent test CLI")
console.log("Connect to a backend, sign in, then send test messages to /api/agent.\n")
console.log("Connect to a backend, sign in, then send test messages to /api/agent/ws.\n")
}
function printUsage(): void {
@@ -348,6 +566,7 @@ function printHelp(): void {
console.log(" /tool Execute an agent debug tool with JSON params")
console.log(" /actions List source actions: /actions <source-id>")
console.log(" /action Execute source action: /action <source-id> <action-id> <json-params>")
console.log(" /conversation Show the current websocket conversation")
console.log(" /session Show the signed-in user")
console.log(" /help Show commands")
console.log(" /quit Exit\n")
@@ -417,6 +636,33 @@ function normalizeBackendUrl(value: string): string {
}
}
function agentWebSocketUrl(backendUrl: string): string {
const url = new URL(backendUrl)
url.protocol = url.protocol === "https:" ? "wss:" : "ws:"
url.pathname = "/api/agent/ws"
url.search = ""
url.hash = ""
return url.toString()
}
function createWebSocketOptions(cookieHeader?: string): Bun.WebSocketOptions | undefined {
if (!cookieHeader) return undefined
return {
headers: {
Cookie: cookieHeader,
},
}
}
function formatWebSocketClose(
url: string,
event: { code: number; reason: string; wasClean: boolean },
): string {
const reason = event.reason ? `: ${event.reason}` : ""
return `Could not connect to ${url} (${event.code}${reason})`
}
function formatPromptLabel(label: string, defaultValue?: string): string {
return defaultValue ? `${label} (${defaultValue}): ` : `${label}: `
}
@@ -511,6 +757,25 @@ function splitSetCookieHeader(header: string): string[] {
return parts.filter(Boolean)
}
function parseJrpcMessage(message: unknown): JrpcMessage | null {
const text = webSocketMessageText(message)
if (!text) return null
try {
const value: unknown = JSON.parse(text)
return isJrpcMessage(value) ? value : null
} catch {
return null
}
}
function webSocketMessageText(message: unknown): string | null {
if (typeof message === "string") return message
if (message instanceof Uint8Array) return Buffer.from(message).toString("utf8")
if (message instanceof ArrayBuffer) return Buffer.from(message).toString("utf8")
return null
}
async function readResponseError(response: Response, path: string): Promise<string> {
const text = await response.text()
if (response.status === 404 && path === "/api/agent") {
@@ -548,11 +813,6 @@ function isAuthSession(value: unknown): value is AuthSession {
)
}
function isQueryResponse(value: unknown): value is QueryResponse {
if (!isJsonObject(value)) return false
return typeof value.message === "string"
}
function isQueryToolsResponse(value: unknown): value is QueryToolsResponse {
if (!isJsonObject(value) || !Array.isArray(value.tools)) return false
return value.tools.every(isQueryToolDefinition)
@@ -585,6 +845,33 @@ function isSourceActionDefinition(value: unknown): value is { id: string; descri
)
}
function isJrpcMessage(value: unknown): value is JrpcMessage {
if (!isJsonObject(value) || value.jsonrpc !== "2.0" || typeof value.id !== "number") {
return false
}
if ("method" in value) {
return (
typeof value.method === "string" &&
(value.params === undefined || Array.isArray(value.params))
)
}
if ("result" in value) {
return true
}
if ("error" in value) {
return isJsonRpcErrorObject(value.error)
}
return false
}
function isJsonRpcErrorObject(value: unknown): boolean {
return isJsonObject(value) && typeof value.code === "number" && typeof value.message === "string"
}
function isJsonObject(value: unknown): value is JsonObject {
return typeof value === "object" && value !== null && !Array.isArray(value)
}

View File

@@ -0,0 +1,11 @@
export class ConversationNotFoundError extends Error {
readonly conversationId: string
readonly userId: string
constructor(conversationId: string, userId: string) {
super(`Conversation "${conversationId}" not found for user "${userId}"`)
this.name = "ConversationNotFoundError"
this.conversationId = conversationId
this.userId = userId
}
}

View File

@@ -2,20 +2,54 @@ import { beforeEach, describe, expect, mock, test } from "bun:test"
import { Hono } from "hono"
import type { Database } from "../db/index.ts"
import type { ConversationRow } from "./storage.ts"
import type {
ConversationEntryRow,
ConversationRow,
ListConversationEntriesParams,
} from "./storage.ts"
import { mockAuthSessionMiddleware } from "../auth/session-middleware.ts"
import { ConversationNotFoundError } from "./errors.ts"
import { registerConversationsHttpHandlers } from "./http.ts"
import { ConversationEntryKind, ConversationEntryVisibility } from "./types.ts"
const MockUserId = "k7Gx2mPqRvNwYs9TdLfA4bHcJeUo1iZn"
const ConversationId = "11111111-1111-4111-8111-111111111111"
const MissingConversationId = "22222222-2222-4222-8222-222222222222"
const conversationRowsByUser = new Map<string, ConversationRow[]>()
const conversationEntryRowsByUserAndConversation = new Map<string, ConversationEntryRow[]>()
const listEntriesCalls: Array<{
userId: string
conversationId: string
params: ListConversationEntriesParams
}> = []
mock.module("./storage.ts", () => ({
conversations: (_db: Database, userId: string) => ({
async listConversations(): Promise<ConversationRow[]> {
return conversationRowsByUser.get(userId) ?? []
},
async listEntries(
conversationId: string,
params: ListConversationEntriesParams = {},
): Promise<ConversationEntryRow[]> {
listEntriesCalls.push({ userId, conversationId, params })
const rows = conversationEntryRowsByUserAndConversation.get(
conversationEntriesKey(userId, conversationId),
)
if (!rows) {
throw new ConversationNotFoundError(conversationId, userId)
}
if (params.visibility) {
return rows.filter((row) => row.visibility === params.visibility)
}
return rows
},
}),
}))
@@ -44,9 +78,39 @@ function createConversationRow(
}
}
function createConversationEntryRow(
id: string,
conversationId: string,
sequence: number,
kind: ConversationEntryRow["kind"],
visibility: ConversationEntryRow["visibility"],
payload: ConversationEntryRow["payload"],
createdAt: string,
metadata: ConversationEntryRow["metadata"] = {},
fileId: string | null = null,
): ConversationEntryRow {
return {
id,
conversationId,
sequence,
kind,
visibility,
fileId,
payload,
metadata,
createdAt: new Date(createdAt),
}
}
function conversationEntriesKey(userId: string, conversationId: string): string {
return `${userId}:${conversationId}`
}
describe("GET /api/conversations", () => {
beforeEach(() => {
conversationRowsByUser.clear()
conversationEntryRowsByUserAndConversation.clear()
listEntriesCalls.length = 0
})
test("returns 401 without auth", async () => {
@@ -108,3 +172,162 @@ describe("GET /api/conversations", () => {
})
})
})
describe("GET /api/conversations/:id/entries", () => {
beforeEach(() => {
conversationRowsByUser.clear()
conversationEntryRowsByUserAndConversation.clear()
listEntriesCalls.length = 0
})
test("returns 401 without auth", async () => {
const app = buildTestApp()
const res = await app.request("/api/conversations/conversation-1/entries")
expect(res.status).toBe(401)
})
test("returns user-visible entries for the authenticated user", async () => {
conversationEntryRowsByUserAndConversation.set(
conversationEntriesKey(MockUserId, ConversationId),
[
createConversationEntryRow(
"entry-user",
ConversationId,
1,
ConversationEntryKind.UserMessage,
ConversationEntryVisibility.UserVisible,
{
role: "user",
parts: [{ type: "text", text: "What is on today?" }],
},
"2026-06-17T09:30:00.000Z",
),
createConversationEntryRow(
"entry-tool",
ConversationId,
2,
ConversationEntryKind.ToolCall,
ConversationEntryVisibility.Internal,
{
toolName: "freya_list_context",
input: {},
},
"2026-06-17T09:30:01.000Z",
),
createConversationEntryRow(
"entry-assistant",
ConversationId,
3,
ConversationEntryKind.AssistantMessage,
ConversationEntryVisibility.UserVisible,
{
role: "assistant",
parts: [{ type: "text", text: "You have two calendar events." }],
},
"2026-06-17T09:30:02.000Z",
{ runId: "run-1" },
),
],
)
const app = buildTestApp("user-1")
const res = await app.request(`/api/conversations/${ConversationId}/entries`)
expect(res.status).toBe(200)
expect(listEntriesCalls).toEqual([
{
userId: MockUserId,
conversationId: ConversationId,
params: { visibility: ConversationEntryVisibility.UserVisible },
},
])
const body = (await res.json()) as { entries: unknown[] }
expect(body).toEqual({
entries: [
{
id: "entry-user",
conversationId: ConversationId,
sequence: 1,
kind: ConversationEntryKind.UserMessage,
visibility: ConversationEntryVisibility.UserVisible,
fileId: null,
payload: {
role: "user",
parts: [{ type: "text", text: "What is on today?" }],
},
metadata: {},
createdAt: "2026-06-17T09:30:00.000Z",
},
{
id: "entry-assistant",
conversationId: ConversationId,
sequence: 3,
kind: ConversationEntryKind.AssistantMessage,
visibility: ConversationEntryVisibility.UserVisible,
fileId: null,
payload: {
role: "assistant",
parts: [{ type: "text", text: "You have two calendar events." }],
},
metadata: { runId: "run-1" },
createdAt: "2026-06-17T09:30:02.000Z",
},
],
})
})
test("returns an empty list when the conversation has no user-visible entries", async () => {
conversationEntryRowsByUserAndConversation.set(
conversationEntriesKey(MockUserId, ConversationId),
[
createConversationEntryRow(
"entry-tool",
ConversationId,
1,
ConversationEntryKind.ToolResult,
ConversationEntryVisibility.Internal,
{ toolCallId: "call-1", output: { ok: true } },
"2026-06-17T09:30:00.000Z",
),
],
)
const app = buildTestApp("user-1")
const res = await app.request(`/api/conversations/${ConversationId}/entries`)
expect(res.status).toBe(200)
const body = (await res.json()) as { entries: unknown[] }
expect(body).toEqual({ entries: [] })
})
test("returns 404 for malformed conversation ids without querying storage", async () => {
const app = buildTestApp("user-1")
const res = await app.request("/api/conversations/missing-conversation/entries")
expect(res.status).toBe(404)
expect(listEntriesCalls).toEqual([])
const body = (await res.json()) as { error: string }
expect(body).toEqual({ error: "Conversation not found" })
})
test("returns 404 when the conversation does not exist for the user", async () => {
const app = buildTestApp("user-1")
const res = await app.request(`/api/conversations/${MissingConversationId}/entries`)
expect(res.status).toBe(404)
expect(listEntriesCalls).toEqual([
{
userId: MockUserId,
conversationId: MissingConversationId,
params: { visibility: ConversationEntryVisibility.UserVisible },
},
])
const body = (await res.json()) as { error: string }
expect(body).toEqual({ error: "Conversation not found" })
})
})

View File

@@ -1,11 +1,15 @@
import type { Context, Hono } from "hono"
import { type } from "arktype"
import { createMiddleware } from "hono/factory"
import type { AuthSessionMiddleware } from "../auth/session-middleware.ts"
import type { Database } from "../db/index.ts"
import type { ConversationRow } from "./storage.ts"
import { ConversationNotFoundError } from "./errors.ts"
import { conversations } from "./storage.ts"
import { ConversationEntryVisibility } from "./types.ts"
type Env = {
Variables: {
@@ -13,11 +17,19 @@ type Env = {
}
}
interface ConversationSummaryResponse {
id: string
createdAt: string
updatedAt: string
}
interface ConversationsHttpHandlersDeps {
db: Database
authSessionMiddleware: AuthSessionMiddleware
}
const ConversationIdParam = type("string.uuid")
export function registerConversationsHttpHandlers(
app: Hono,
{ db, authSessionMiddleware }: ConversationsHttpHandlersDeps,
@@ -28,6 +40,7 @@ export function registerConversationsHttpHandlers(
})
app.get("/api/conversations", inject, authSessionMiddleware, handleListConversations)
app.get("/api/conversations/:id/entries", inject, authSessionMiddleware, handleListEntries)
}
async function handleListConversations(c: Context<Env>) {
@@ -35,10 +48,54 @@ async function handleListConversations(c: Context<Env>) {
const db = c.get("db")
return c.json({
conversations: (await conversations(db, user.id).listConversations()).map((row) => ({
conversations: (await conversations(db, user.id).listConversations()).map(
serializeConversation,
),
})
}
async function handleListEntries(c: Context<Env>) {
const user = c.get("user")!
const db = c.get("db")
const conversationId = c.req.param("id")
if (!conversationId) {
return c.json({ error: "Conversation not found" }, 404)
}
const parsedConversationId = ConversationIdParam(conversationId)
if (parsedConversationId instanceof type.errors) {
return c.json({ error: "Conversation not found" }, 404)
}
try {
const entries = await conversations(db, user.id).listEntries(parsedConversationId, {
visibility: ConversationEntryVisibility.UserVisible,
})
return c.json({
entries: entries.map((row) => ({
id: row.id,
conversationId: row.conversationId,
sequence: row.sequence,
kind: row.kind,
visibility: row.visibility,
fileId: row.fileId,
payload: row.payload,
metadata: row.metadata,
createdAt: row.createdAt.toISOString(),
})),
})
} catch (err) {
if (err instanceof ConversationNotFoundError) {
return c.json({ error: "Conversation not found" }, 404)
}
throw err
}
}
function serializeConversation(row: ConversationRow): ConversationSummaryResponse {
return {
id: row.id,
createdAt: row.createdAt.toISOString(),
updatedAt: row.updatedAt.toISOString(),
})),
})
}
}

View File

@@ -19,6 +19,7 @@ import {
files,
user,
} from "../db/schema.ts"
import { ConversationNotFoundError } from "./errors.ts"
import {
ConversationEntryMetadata as ConversationEntryMetadataSchema,
AssistantMessagePayload as AssistantMessagePayloadSchema,
@@ -96,7 +97,7 @@ export interface ListConversationEntriesParams {
}
export function conversations(db: Database, userId: string) {
return {
const storage = {
async createConversation(): Promise<ConversationRow> {
return insertConversation(db, userId)
},
@@ -109,6 +110,18 @@ export function conversations(db: Database, userId: string) {
.orderBy(desc(conversationsTable.updatedAt), desc(conversationsTable.createdAt))
},
async getConversation(conversationId: string): Promise<ConversationRow | null> {
const rows = await db
.select()
.from(conversationsTable)
.where(
and(eq(conversationsTable.id, conversationId), eq(conversationsTable.userId, userId)),
)
.limit(1)
return rows[0] ?? null
},
async getOrCreateConversation(): Promise<ConversationRow> {
return db.transaction(async (tx) => {
await requireUserForUpdate(tx, userId)
@@ -141,7 +154,9 @@ export function conversations(db: Database, userId: string) {
}
const rows = await db.transaction(async (tx) => {
await requireConversationForUpdate(tx, userId, conversationId)
if (!(await findConversationForUpdate(tx, userId, conversationId))) {
throw new ConversationNotFoundError(conversationId, userId)
}
const sequence = await nextSequence(tx, conversationId)
const rows = await tx
@@ -175,7 +190,9 @@ export function conversations(db: Database, userId: string) {
const metadata = ConversationEntryMetadataSchema.assert(input.metadata ?? {})
return db.transaction(async (tx) => {
await requireConversationForUpdate(tx, userId, conversationId)
if (!(await findConversationForUpdate(tx, userId, conversationId))) {
throw new ConversationNotFoundError(conversationId, userId)
}
const file = await insertFile(tx, userId, input.file)
const sequence = await nextSequence(tx, conversationId)
@@ -204,7 +221,9 @@ export function conversations(db: Database, userId: string) {
conversationId: string,
params: ListConversationEntriesParams = {},
): Promise<ConversationEntryRow[]> {
await requireConversation(db, userId, conversationId)
if (!(await storage.getConversation(conversationId))) {
throw new ConversationNotFoundError(conversationId, userId)
}
if (params.visibility) {
return db
@@ -226,6 +245,8 @@ export function conversations(db: Database, userId: string) {
.orderBy(asc(conversationEntries.sequence))
},
}
return storage
}
function payloadForKind(
@@ -259,25 +280,11 @@ async function requireUserForUpdate(db: Database, userId: string): Promise<void>
requireRow(rows, `User not found: ${userId}`)
}
async function requireConversation(
async function findConversationForUpdate(
db: Database,
userId: string,
conversationId: string,
): Promise<ConversationRow> {
const rows = await db
.select()
.from(conversationsTable)
.where(and(eq(conversationsTable.id, conversationId), eq(conversationsTable.userId, userId)))
.limit(1)
return requireRow(rows, `Conversation not found: ${conversationId}`)
}
async function requireConversationForUpdate(
db: Database,
userId: string,
conversationId: string,
): Promise<ConversationRow> {
): Promise<ConversationRow | null> {
const rows = await db
.select()
.from(conversationsTable)
@@ -285,7 +292,7 @@ async function requireConversationForUpdate(
.limit(1)
.for("update")
return requireRow(rows, `Conversation not found: ${conversationId}`)
return rows[0] ?? null
}
async function latestConversation(db: Database, userId: string): Promise<ConversationRow | null> {

View File

@@ -11,6 +11,7 @@ import { registerAuthHandlers } from "./auth/http.ts"
import { createAuth } from "./auth/index.ts"
import { createRequireSession } from "./auth/session-middleware.ts"
import { CalDavSourceProvider } from "./caldav/provider.ts"
import { registerConversationsHttpHandlers } from "./conversations/http.ts"
import { createDatabase } from "./db/index.ts"
import { registerFeedHttpHandlers } from "./engine/http.ts"
import { createFeedEnhancer } from "./enhancement/enhance-feed.ts"
@@ -129,6 +130,7 @@ function main() {
sessionManager,
authSessionMiddleware,
})
registerConversationsHttpHandlers(app, { db, authSessionMiddleware })
if (isDebugMode) {
registerDebugAgentHttpHandlers(app, {
authSessionMiddleware,

View File

@@ -49,6 +49,10 @@
"apps/agent-test-cli": {
"name": "@freya/agent-test-cli",
"version": "0.0.0",
"dependencies": {
"@freya/agent-protocol": "workspace:*",
"@nym.sh/jrpc": "^0.1.0",
},
},
"apps/freya-backend": {
"name": "@freya/backend",

27
flake.lock generated Normal file
View File

@@ -0,0 +1,27 @@
{
"nodes": {
"nixpkgs": {
"locked": {
"lastModified": 1781577229,
"narHash": "sha256-lrp67w8AulE9Ks53n27I45ADSzbOCn4H+CNW1Ck8B+8=",
"owner": "NixOS",
"repo": "nixpkgs",
"rev": "567a49d1913ce81ac6e9582e3553dd90a955875f",
"type": "github"
},
"original": {
"owner": "NixOS",
"ref": "nixos-unstable",
"repo": "nixpkgs",
"type": "github"
}
},
"root": {
"inputs": {
"nixpkgs": "nixpkgs"
}
}
},
"root": "root",
"version": 7
}

280
flake.nix Normal file
View File

@@ -0,0 +1,280 @@
{
description = "FREYA development shell";
inputs = {
nixpkgs.url = "github:NixOS/nixpkgs/nixos-unstable";
};
outputs =
{ nixpkgs, ... }:
let
systems = [
"x86_64-linux"
"aarch64-linux"
"x86_64-darwin"
"aarch64-darwin"
];
lib = nixpkgs.lib;
forEachSystem = lib.genAttrs systems;
pkgsFor = forEachSystem (system: import nixpkgs { inherit system; });
# App outputs are for long-running local tools and dev servers.
appScripts = {
expo = "expo";
drizzle-studio = "drizzle-studio";
freya-backend = "freya-backend";
admin-dashboard = "admin-dashboard";
agent-test-cli = "agent-test-cli";
};
# Check outputs are the CI-like validation commands run by `nix flake check`.
checkCommands = {
format-check = "bun run format:check";
lint = "bun run lint";
test = "bun run test";
};
# Dev-shell conveniences mirror the common app/check commands.
shellScripts = appScripts // {
freya-test = "test";
lint = "lint";
format-check = "format:check";
};
# node_modules is content-addressed. If bun.lock or package manifests
# change, Nix will report the new hash to put here.
nodeModulesHashes = {
x86_64-linux = "sha256-apVZaFGf9OKpil1WdcQ1CJODsIdjLWlBBZErHg5mjZA=";
};
checkSystems = lib.attrNames nodeModulesHashes;
# Dependency derivations only need the lockfile and workspace manifests,
# so source-only edits do not force Bun to reinstall.
dependencySource = lib.fileset.toSource {
root = ./.;
fileset = lib.fileset.fileFilter (file: file.name == "bun.lock" || file.name == "package.json") ./.;
};
# Checks run against a clean source tree, even when using `path:.`.
# Without this filter, local node_modules can sneak into the Nix sandbox.
projectSource = builtins.path {
name = "freya-source";
path = ./.;
filter =
path: type:
let
name = builtins.baseNameOf path;
in
!(type == "directory" && (name == ".git" || name == "node_modules")) && name != "result";
};
mkBunScriptCommands =
pkgs: scripts:
let
mkBunScript =
name: script:
pkgs.writeShellApplication {
inherit name;
runtimeInputs = with pkgs; [
bun
git
];
text = ''
repo_root="$(git rev-parse --show-toplevel 2>/dev/null || pwd)"
cd "$repo_root"
exec bun run ${lib.escapeShellArg script} "$@"
'';
};
in
lib.mapAttrs mkBunScript scripts;
mkBunApps =
commands:
lib.mapAttrs (name: command: {
type = "app";
program = "${command}/bin/${name}";
}) commands;
mkBunNodeModules =
system: pkgs:
pkgs.stdenvNoCC.mkDerivation {
pname = "freya-node-modules";
version = "1";
__structuredAttrs = true;
src = dependencySource;
nativeBuildInputs = with pkgs; [
bun
cacert
nodejs
];
SSL_CERT_FILE = "${pkgs.cacert}/etc/ssl/certs/ca-bundle.crt";
GIT_SSL_CAINFO = "${pkgs.cacert}/etc/ssl/certs/ca-bundle.crt";
outputHashAlgo = "sha256";
outputHashMode = "recursive";
outputHash = nodeModulesHashes.${system};
# `patchShebangs` embeds Nix store interpreters in package bins. The
# check derivations also depend on bun/node, so this dependency blob
# can safely drop those references after its hash is verified.
unsafeDiscardReferences.out = true;
dontConfigure = true;
# Workspace package links are completed inside each check's source tree,
# so they are intentionally dangling in this dependency-only output.
dontFixup = true;
buildPhase = ''
runHook preBuild
export HOME="$TMPDIR/home"
mkdir -p "$HOME"
# Keep the real workspace manifest for `--frozen-lockfile`, but
# filter out frontend workspaces that do not participate in checks.
# `--force` matters in the Nix sandbox: without it, Bun can accept
# manifest-only cached packages and leave tool binaries missing.
bun install \
--force \
--frozen-lockfile \
--ignore-scripts \
--backend copyfile \
--filter freya \
--filter '@freya/*' \
--filter '@freya/backend' \
--no-progress
patchShebangs node_modules
runHook postBuild
'';
installPhase = ''
runHook preInstall
mkdir -p "$out"
# Keep the root install in the store; checks symlink this directly.
cp -a node_modules "$out/node_modules"
# Bun also creates per-workspace node_modules directories. These are
# mostly relative symlinks, so checks copy the symlink entries into
# their writable source tree instead of symlinking the directory.
find apps packages -mindepth 2 -maxdepth 2 -type d -name node_modules -print |
while IFS= read -r node_modules_dir; do
mkdir -p "$out/$(dirname "$node_modules_dir")"
cp -a "$node_modules_dir" "$out/$node_modules_dir"
done
runHook postInstall
'';
};
mkBunCheck =
pkgs: nodeModules: name: command:
pkgs.stdenvNoCC.mkDerivation {
pname = "freya-${name}";
version = "1";
src = projectSource;
nativeBuildInputs = with pkgs; [
bun
nodejs
];
dontConfigure = true;
buildPhase = ''
runHook preBuild
export HOME="$TMPDIR/home"
mkdir -p "$HOME"
# Root dependencies are read-only and shared across checks.
ln -s "${nodeModules}/node_modules" node_modules
# Workspace node_modules contain relative symlinks back to packages/
# and apps/, so copy just those symlink entries into this source tree.
for node_modules_dir in "${nodeModules}"/apps/*/node_modules "${nodeModules}"/packages/*/node_modules; do
if [ -d "$node_modules_dir" ]; then
relative_path="''${node_modules_dir#"${nodeModules}/"}"
mkdir -p "$relative_path"
cp -a "$node_modules_dir/." "$relative_path/"
fi
done
${command}
runHook postBuild
'';
installPhase = ''
runHook preInstall
mkdir -p "$out"
touch "$out/${name}"
runHook postInstall
'';
};
in
{
apps = forEachSystem (
system:
let
pkgs = pkgsFor.${system};
in
mkBunApps (mkBunScriptCommands pkgs appScripts)
);
checks = lib.genAttrs checkSystems (
system:
let
pkgs = pkgsFor.${system};
nodeModules = mkBunNodeModules system pkgs;
in
lib.mapAttrs (mkBunCheck pkgs nodeModules) checkCommands
);
devShells = forEachSystem (
system:
let
pkgs = pkgsFor.${system};
bunScriptCommands = lib.attrValues (mkBunScriptCommands pkgs shellScripts);
commonPackages = with pkgs; [
bun
eas-cli
git
gh
gnumake
nixfmt
nodejs
openssl
pkg-config
postgresql
python3
watchman
];
linuxPackages = with pkgs; [
gcc
inotify-tools
tailscale
];
in
{
default = pkgs.mkShell {
packages =
commonPackages ++ bunScriptCommands ++ pkgs.lib.optionals pkgs.stdenv.isLinux linuxPackages;
SSL_CERT_FILE = "${pkgs.cacert}/etc/ssl/certs/ca-bundle.crt";
shellHook = ''
export PATH="$PWD/node_modules/.bin:$PATH"
'';
};
}
);
formatter = forEachSystem (system: pkgsFor.${system}.nixfmt);
};
}

View File

@@ -1,4 +1,4 @@
import { describe, expect, test } from "bun:test"
import { describe, expect, spyOn, test } from "bun:test"
import type { ActionDefinition, ContextEntry, ContextKey, FeedItem, FeedSource } from "./index"
@@ -145,6 +145,16 @@ function createAlertSource(): FeedSource<AlertFeedItem> {
}
}
async function waitForCondition(predicate: () => boolean, timeoutMs = 2_000): Promise<void> {
const deadline = Date.now() + timeoutMs
while (!predicate()) {
if (Date.now() > deadline) {
throw new Error("Timed out waiting for condition")
}
await new Promise((resolve) => setTimeout(resolve, 10))
}
}
// =============================================================================
// TESTS
// =============================================================================
@@ -807,11 +817,14 @@ describe("FeedEngine", () => {
})
test("TTL resets after reactive update", async () => {
let now = 1_000
const nowSpy = spyOn(Date, "now").mockImplementation(() => now)
const location = createLocationSource()
const weather = createWeatherSource()
const engine = new FeedEngine({ cacheTtlMs: 100 }).register(location).register(weather)
try {
engine.start()
// Initial reactive update
@@ -820,15 +833,19 @@ describe("FeedEngine", () => {
expect(engine.lastFeed()).not.toBeNull()
// Wait 70ms (total 120ms from first update, past original TTL)
// but trigger another update at 50ms to reset TTL
// Move past the original TTL, then trigger another update to reset it.
now += 120
location.simulateUpdate({ lat: 52.0, lng: -0.2 })
await new Promise((resolve) => setTimeout(resolve, 50))
// Should still be cached because TTL was reset by second update
// Should still be cached because TTL was reset by second update.
expect(engine.lastFeed()).not.toBeNull()
engine.stop()
} finally {
engine.stop()
nowSpy.mockRestore()
}
})
test("cacheTtlMs is configurable", async () => {
@@ -869,17 +886,21 @@ describe("FeedEngine", () => {
},
}
const engine = new FeedEngine({ cacheTtlMs: 50 }).register(source)
const engine = new FeedEngine({ cacheTtlMs: 20 }).register(source)
await engine.refresh()
expect(fetchCount).toBe(1)
try {
engine.start()
// Wait for two TTL intervals to elapse
await new Promise((resolve) => setTimeout(resolve, 120))
await waitForCondition(() => fetchCount >= 2)
// Should have auto-refreshed at least twice
expect(fetchCount).toBeGreaterThanOrEqual(2)
expect(engine.lastFeed()).not.toBeNull()
} finally {
engine.stop()
}
})
test("stop cancels periodic refresh", async () => {
@@ -935,28 +956,25 @@ describe("FeedEngine", () => {
},
}
const engine = new FeedEngine({ cacheTtlMs: 100 })
const engine = new FeedEngine({ cacheTtlMs: 10_000 })
.register(location)
.register(countingWeather)
const clearTimeoutSpy = spyOn(globalThis, "clearTimeout")
try {
engine.start()
// At 40ms, push a reactive update — this resets the timer
await new Promise((resolve) => setTimeout(resolve, 40))
const countBeforeUpdate = fetchCount
location.simulateUpdate({ lat: 51.5, lng: -0.1 })
await new Promise((resolve) => setTimeout(resolve, 20))
await waitForCondition(() => fetchCount > countBeforeUpdate && engine.lastFeed() !== null)
// Reactive update triggered a fetch
// Reactive updates refresh the cache and reset the pending periodic timer.
expect(fetchCount).toBeGreaterThan(countBeforeUpdate)
const countAfterUpdate = fetchCount
// At 100ms from start (60ms after reactive update), the original
// timer would have fired, but it was reset. No extra fetch yet.
await new Promise((resolve) => setTimeout(resolve, 40))
expect(fetchCount).toBe(countAfterUpdate)
expect(clearTimeoutSpy).toHaveBeenCalled()
} finally {
engine.stop()
clearTimeoutSpy.mockRestore()
}
})
})