328 lines
18 KiB
Swift
328 lines
18 KiB
Swift
|
import Foundation
|
||
|
|
import Testing
|
||
|
|
@testable import Sybil
|
||
|
|
|
||
|
|
private func agentMessage(_ id: String, entry: String, role: MessageRole = .assistant, info: [String: JSONValue] = [:]) -> Message {
|
||
|
|
Message(id: id, createdAt: Date(timeIntervalSince1970: 1_700_000_000), role: role, content: "Message \(id)", metadata: .object(["kind": .string(role == .tool ? "tool_call" : "message"), "entryId": .string(entry)].merging(info, uniquingKeysWith: { _, b in b })))
|
||
|
|
}
|
||
|
|
private func agentPage(_ messages: [Message], cursor: String, more: Bool = false) -> AgentPage {
|
||
|
|
AgentPage(conversationId: "root", messages: messages, cursor: cursor, beforeCursor: nil, hasMore: more)
|
||
|
|
}
|
||
|
|
|
||
|
|
private final class MemoryOutbox: @unchecked Sendable {
|
||
|
|
private let lock = NSLock()
|
||
|
|
private var rows: [AgentOutboxRow] = []
|
||
|
|
var storage: AgentOutboxStorage {
|
||
|
|
AgentOutboxStorage(read: { self.read() }, write: { self.save($0) })
|
||
|
|
}
|
||
|
|
func read() -> [AgentOutboxRow] { lock.withLock { rows } }
|
||
|
|
func save(_ value: [AgentOutboxRow]) { lock.withLock { rows = value } }
|
||
|
|
}
|
||
|
|
private actor ControlledAgentSocket: AgentSocket {
|
||
|
|
let stream: AsyncThrowingStream<Data, any Error>
|
||
|
|
let continuation: AsyncThrowingStream<Data, any Error>.Continuation
|
||
|
|
private(set) var closed = false
|
||
|
|
init() { (stream, continuation) = AsyncThrowingStream.makeStream() }
|
||
|
|
func receive() async throws -> Data {
|
||
|
|
var iterator = stream.makeAsyncIterator()
|
||
|
|
guard let data = try await iterator.next() else { throw CancellationError() }
|
||
|
|
return data
|
||
|
|
}
|
||
|
|
func send(_ text: String) {}
|
||
|
|
func close() { closed = true; continuation.finish(throwing: URLError(.networkConnectionLost)) }
|
||
|
|
func closeCode() -> Int { 0 }
|
||
|
|
func push(_ text: String) { continuation.yield(Data(text.utf8)) }
|
||
|
|
}
|
||
|
|
private actor TestAgentClient: AgentAPIClienting {
|
||
|
|
var pages: [String: AgentPage]
|
||
|
|
var requests: [String] = []
|
||
|
|
var submissions: [(String, String)] = []
|
||
|
|
var read = "0"
|
||
|
|
var failRead = false
|
||
|
|
var failSubmit = false
|
||
|
|
var malformedReceipt = false
|
||
|
|
var socketAfters: [String] = []
|
||
|
|
var sockets: [ControlledAgentSocket] = []
|
||
|
|
var holdRead = false
|
||
|
|
var heldRead: CheckedContinuation<AgentReadState, Never>?
|
||
|
|
var beforeSubmit: (@Sendable () -> Bool)?
|
||
|
|
init(pages: [String: AgentPage] = [:]) { self.pages = pages }
|
||
|
|
func agentDescriptor() -> AgentDescriptor {
|
||
|
|
AgentDescriptor(id: "root", title: "Sybil Agent", type: "persistent-agent", model: "test", protocol: "pi-room.v1", activity: AgentActivity(), websocket: "/v1/agent/ws")
|
||
|
|
}
|
||
|
|
func agentMessages(after: String?, before: String?) -> AgentPage {
|
||
|
|
let cursor = after ?? "0"; requests.append(cursor)
|
||
|
|
return pages[cursor] ?? agentPage([], cursor: cursor)
|
||
|
|
}
|
||
|
|
func agentReadState() async -> AgentReadState {
|
||
|
|
if holdRead { return await withCheckedContinuation { heldRead = $0 } }
|
||
|
|
return AgentReadState(cursor: read)
|
||
|
|
}
|
||
|
|
func pauseRead() { holdRead = true }
|
||
|
|
func releaseRead() { holdRead = false; heldRead?.resume(returning: AgentReadState(cursor: read)); heldRead = nil }
|
||
|
|
func hasHeldRead() -> Bool { heldRead != nil }
|
||
|
|
func agentMarkRead(cursor: String) throws -> AgentReadState {
|
||
|
|
if failRead { throw AgentFailure(message: "Offline") }
|
||
|
|
read = try AgentCursor.advance(read, cursor)
|
||
|
|
return AgentReadState(cursor: read)
|
||
|
|
}
|
||
|
|
func agentSubmit(content: String, requestID: String) async throws -> AgentReceipt {
|
||
|
|
#expect(beforeSubmit?() ?? true)
|
||
|
|
submissions.append((content, requestID))
|
||
|
|
try await Task.sleep(for: .milliseconds(80))
|
||
|
|
if failSubmit { throw AgentFailure(message: "Lost acknowledgement") }
|
||
|
|
return AgentReceipt(id: 1, status: "queued", messageId: malformedReceipt ? "wrong" : "inbound:sybil:\(requestID)")
|
||
|
|
}
|
||
|
|
func agentSocket(after: String) async -> any AgentSocket {
|
||
|
|
socketAfters.append(after)
|
||
|
|
let socket = ControlledAgentSocket(); sockets.append(socket)
|
||
|
|
await socket.push(#"{"type":"ready","conversationId":"root"}"#)
|
||
|
|
return socket
|
||
|
|
}
|
||
|
|
func connectedAfters() -> [String] { socketAfters }
|
||
|
|
func latestSocket() -> ControlledAgentSocket? { sockets.last }
|
||
|
|
func setFailures(submit: Bool = false, read: Bool = false, malformed: Bool = false) { failSubmit = submit; failRead = read; malformedReceipt = malformed }
|
||
|
|
func requirePersisted(_ check: @escaping @Sendable () -> Bool) { beforeSubmit = check }
|
||
|
|
func submittedIDs() -> [String] { submissions.map(\.1) }
|
||
|
|
func requestedCursors() -> [String] { requests }
|
||
|
|
}
|
||
|
|
|
||
|
|
@MainActor
|
||
|
|
private func eventually(_ predicate: () -> Bool) async -> Bool {
|
||
|
|
for _ in 0..<200 {
|
||
|
|
if predicate() { return true }
|
||
|
|
try? await Task.sleep(for: .milliseconds(10))
|
||
|
|
}
|
||
|
|
return false
|
||
|
|
}
|
||
|
|
|
||
|
|
struct AgentTests {
|
||
|
|
@Test func cursorAndMalformedFrames() throws {
|
||
|
|
#expect(try AgentCursor.normalize("00012") == "12")
|
||
|
|
#expect(try AgentCursor.advance("9", "10") == "10")
|
||
|
|
#expect(try AgentCursor.advance("9007199254740993", "9007199254740992") == "9007199254740993")
|
||
|
|
#expect(throws: (any Error).self) { try AgentCursor.normalize("-1") }
|
||
|
|
var state = AgentRoomState()
|
||
|
|
try state.apply(agentPage([agentMessage("reply", entry: "10")], cursor: "10"))
|
||
|
|
let original = state
|
||
|
|
#expect(throws: (any Error).self) { try state.apply(agentPage([], cursor: "bad")) }
|
||
|
|
#expect(state == original)
|
||
|
|
#expect(throws: (any Error).self) { try AgentEvent.decode(Data(#"{"type":"messages","conversationId":"root","cursor":"99","beforeCursor":null,"hasMore":false,"messages":[{"id":"broken"}]}"#.utf8)) }
|
||
|
|
#expect(state.cursor == "10")
|
||
|
|
let oversizedDuration = agentMessage("tool", entry: "11", role: .tool, info: ["durationMs": .number(1e100)])
|
||
|
|
#expect(oversizedDuration.toolCallMetadata?.durationMs == nil)
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test func terminalToolAndRestSocketRace() throws {
|
||
|
|
let start = agentMessage("tool:1", entry: "5", role: .tool, info: ["status": .string("initiated"), "toolName": .string("search"), "args": .object(["query": .string("fixture")]), "startedAt": .string("start")])
|
||
|
|
var end = agentMessage("tool:1", entry: "8", role: .tool, info: ["status": .string("failed"), "error": .string("fixture failure"), "resultPreview": .string("detail"), "args": .object([:])])
|
||
|
|
end.createdAt.addTimeInterval(5)
|
||
|
|
var state = AgentRoomState()
|
||
|
|
try state.apply(agentPage([start], cursor: "5"))
|
||
|
|
try state.apply(agentPage([end, agentMessage("reply", entry: "9")], cursor: "9"))
|
||
|
|
try state.apply(agentPage([start], cursor: "5"))
|
||
|
|
#expect(state.cursor == "9")
|
||
|
|
#expect(state.messages.count == 2)
|
||
|
|
let tool = try #require(state.messages.first?.toolCallMetadata)
|
||
|
|
#expect(tool.status == "failed")
|
||
|
|
#expect(tool.error == "fixture failure")
|
||
|
|
#expect(tool.args == .object(["query": .string("fixture")]))
|
||
|
|
#expect(tool.startedAt == "start")
|
||
|
|
#expect(state.messages.first?.createdAt == start.createdAt)
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test func unreadExcludesToolsAndUsers() throws {
|
||
|
|
var state = AgentRoomState()
|
||
|
|
try state.apply(agentPage([agentMessage("a", entry: "9"), agentMessage("b", entry: "10"), agentMessage("u", entry: "11", role: .user), agentMessage("t", entry: "12", role: .tool)], cursor: "12"))
|
||
|
|
state.readCursor = "9"
|
||
|
|
#expect(state.unreadCount == 1)
|
||
|
|
state.readCursor = try AgentCursor.advance(state.readCursor, "12")
|
||
|
|
state.readCursor = try AgentCursor.advance(state.readCursor, "2")
|
||
|
|
#expect(state.unreadCount == 0)
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test func typingReplyAndStaleActivityAreAtomic() throws {
|
||
|
|
var presentation = AgentPresentation(), state = AgentRoomState()
|
||
|
|
let now = Date(timeIntervalSince1970: 100)
|
||
|
|
var busy = AgentActivity(); busy.isThinking = true; busy.backgroundWorkers = 1
|
||
|
|
var after = state; after.activity = busy
|
||
|
|
presentation.receive(.activity("root", busy), before: state, after: after, replay: false, now: now)
|
||
|
|
state = after
|
||
|
|
#expect(!presentation.isTyping(now: now.addingTimeInterval(0.249)))
|
||
|
|
#expect(presentation.isTyping(now: now.addingTimeInterval(0.251)))
|
||
|
|
let reply = agentPage([agentMessage("reply", entry: "1")], cursor: "1")
|
||
|
|
try after.apply(reply)
|
||
|
|
presentation.receive(.messages(reply), before: state, after: after, replay: false, now: now.addingTimeInterval(1))
|
||
|
|
state = after
|
||
|
|
#expect(!presentation.isTyping(now: now.addingTimeInterval(2)))
|
||
|
|
#expect(state.activity.isBusy)
|
||
|
|
presentation.receive(.activity("root", busy), before: state, after: state, replay: false, now: now.addingTimeInterval(3))
|
||
|
|
#expect(!presentation.isTyping(now: now.addingTimeInterval(4)))
|
||
|
|
for name in AgentPresentation.quietTools {
|
||
|
|
busy.tools.append(.init(id: name, name: name, status: "initiated"))
|
||
|
|
presentation.receive(.activity("root", busy), before: state, after: state, replay: false, now: now.addingTimeInterval(5))
|
||
|
|
#expect(!presentation.isTyping(now: now.addingTimeInterval(6)))
|
||
|
|
}
|
||
|
|
busy.tools.append(.init(id: "real", name: "search", status: "initiated"))
|
||
|
|
presentation.receive(.activity("root", busy), before: state, after: state, replay: false, now: now.addingTimeInterval(7))
|
||
|
|
#expect(presentation.isTyping(now: now.addingTimeInterval(7.3)))
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test func socketURLAndAuthentication() throws {
|
||
|
|
let config = APIConfiguration(baseURL: URL(string: "https://sybil.bajor.cloud/api/?accidental=query")!, authToken: " token ")
|
||
|
|
#expect(try config.agentWebSocketURL().absoluteString == "wss://sybil.bajor.cloud/api/v1/agent/ws")
|
||
|
|
let frame = try JSONDecoder().decode(JSONValue.self, from: Data(config.agentAuthFrame(after: "0009").utf8))
|
||
|
|
#expect(frame.objectValue?["token"] == .string("token"))
|
||
|
|
#expect(frame.objectValue?["after"] == .string("9"))
|
||
|
|
let open = APIConfiguration(baseURL: URL(string: "http://localhost:8787")!, authToken: nil)
|
||
|
|
let openFrame = try JSONDecoder().decode(JSONValue.self, from: Data(open.agentAuthFrame(after: "0").utf8))
|
||
|
|
#expect(openFrame.objectValue?["token"] == .null)
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test @MainActor func emptyPagesAndInitialHistoryDoNotAnimate() async throws {
|
||
|
|
let client = TestAgentClient(pages: ["0": agentPage([], cursor: "3", more: true), "3": agentPage([], cursor: "7", more: true), "7": agentPage([agentMessage("reply", entry: "8")], cursor: "8")])
|
||
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
||
|
|
defer { room.stop() }
|
||
|
|
#expect(await eventually { !room.isLoading })
|
||
|
|
#expect(await client.requestedCursors() == ["0", "3", "7"])
|
||
|
|
#expect(room.snapshot.room.cursor == "8")
|
||
|
|
#expect(room.unreadCount == 1)
|
||
|
|
#expect(room.snapshot.entranceIDs.isEmpty)
|
||
|
|
try room.receive(.messages(agentPage([agentMessage("late", entry: "9")], cursor: "9")))
|
||
|
|
room.flush()
|
||
|
|
#expect(room.snapshot.entranceIDs == ["late"])
|
||
|
|
try room.receive(.messages(agentPage([agentMessage("late", entry: "9")], cursor: "9")))
|
||
|
|
room.flush()
|
||
|
|
#expect(room.snapshot.entranceIDs.isEmpty)
|
||
|
|
#expect(room.messages.count == 2)
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test @MainActor func outboxPersistsBeforePostAndEchoWinsFailure() async throws {
|
||
|
|
let memory = MemoryOutbox(), client = TestAgentClient()
|
||
|
|
await client.setFailures(submit: true)
|
||
|
|
await client.requirePersisted { !memory.read().isEmpty }
|
||
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: memory.storage)
|
||
|
|
defer { room.stop() }
|
||
|
|
#expect(await eventually { !room.isLoading })
|
||
|
|
#expect(room.send("identical")); #expect(room.send("identical"))
|
||
|
|
#expect(room.messages.count == 2)
|
||
|
|
let first = room.outbox[0], second = room.outbox[1]
|
||
|
|
#expect(first.requestID != second.requestID)
|
||
|
|
#expect(memory.read().count == 2)
|
||
|
|
try room.receive(.messages(agentPage([agentMessage(first.id, entry: "1", role: .user)], cursor: "1")))
|
||
|
|
room.flush()
|
||
|
|
#expect(await eventually { room.outbox[1].status == "failed" })
|
||
|
|
#expect(room.outbox[0].status == "complete")
|
||
|
|
#expect(room.messages.first(where: { $0.id == first.id })?.createdAt == first.createdAt)
|
||
|
|
room.retry(second.id)
|
||
|
|
#expect(await eventually { room.outbox[1].status == "failed" })
|
||
|
|
let ids = await client.submittedIDs()
|
||
|
|
#expect(ids.filter { $0 == second.requestID }.count == 2)
|
||
|
|
let restored = SybilAgentRoom(); restored.configure(client: client, storage: memory.storage)
|
||
|
|
defer { restored.stop() }
|
||
|
|
#expect(restored.outbox.count == 1)
|
||
|
|
#expect(restored.outbox[0].requestID == second.requestID)
|
||
|
|
#expect(restored.outbox[0].createdAt == second.createdAt)
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test @MainActor func malformedReceiptAndStorageFailureRemainRetryable() async {
|
||
|
|
let client = TestAgentClient(); await client.setFailures(malformed: true)
|
||
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
||
|
|
defer { room.stop() }
|
||
|
|
#expect(await eventually { !room.isLoading })
|
||
|
|
room.send("fixture")
|
||
|
|
#expect(await eventually { room.outbox.first?.status == "failed" })
|
||
|
|
#expect(room.outbox.first?.seenOnServer == false)
|
||
|
|
let failed = SybilAgentRoom()
|
||
|
|
failed.configure(client: client, storage: AgentOutboxStorage(read: { [] }, write: { _ in throw AgentFailure(message: "Disk full") }))
|
||
|
|
defer { failed.stop() }
|
||
|
|
failed.send("must not submit")
|
||
|
|
#expect(failed.outbox.first?.status == "failed")
|
||
|
|
#expect(await client.submittedIDs().count == 1)
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test @MainActor func readOnlyAfterAcknowledgementAndCapturedSnapshot() async throws {
|
||
|
|
let client = TestAgentClient(pages: ["0": agentPage([agentMessage("a", entry: "1")], cursor: "1")])
|
||
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
||
|
|
defer { room.stop() }
|
||
|
|
#expect(await eventually { !room.isLoading })
|
||
|
|
room.isSelected = true
|
||
|
|
await client.setFailures(read: true)
|
||
|
|
room.markRead(renderedCursor: "1")
|
||
|
|
#expect(await eventually { room.error != nil })
|
||
|
|
#expect(room.snapshot.room.readCursor == "0")
|
||
|
|
await client.setFailures()
|
||
|
|
try room.receive(.messages(agentPage([agentMessage("b", entry: "2")], cursor: "2")))
|
||
|
|
room.markRead(renderedCursor: "1")
|
||
|
|
#expect(await eventually { room.snapshot.room.readCursor == "1" })
|
||
|
|
#expect(room.unreadCount == 1)
|
||
|
|
room.isObscured = true; room.markRead(renderedCursor: "2")
|
||
|
|
try? await Task.sleep(for: .milliseconds(100))
|
||
|
|
#expect(room.snapshot.room.readCursor == "1")
|
||
|
|
try room.applyRead(AgentReadState(cursor: "2")); room.flush()
|
||
|
|
#expect(room.unreadCount == 0)
|
||
|
|
room.setActive(false); room.setActive(true)
|
||
|
|
#expect(await eventually { room.connection == "Connecting" })
|
||
|
|
#expect(room.messages.count == 2)
|
||
|
|
}
|
||
|
|
@Test @MainActor func interruptedAdmissionForegroundKeepsRetryIdentity() async throws {
|
||
|
|
let client = TestAgentClient(), memory = MemoryOutbox()
|
||
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: memory.storage)
|
||
|
|
defer { room.stop() }
|
||
|
|
#expect(await eventually { room.connection == "Connected" })
|
||
|
|
room.send("interrupted fixture")
|
||
|
|
let original = try #require(room.outbox.first)
|
||
|
|
room.setActive(false)
|
||
|
|
#expect(room.outbox.first?.status == "failed")
|
||
|
|
#expect(memory.read().first?.requestID == original.requestID)
|
||
|
|
room.setActive(true)
|
||
|
|
#expect(await eventually { room.connection == "Connected" })
|
||
|
|
room.retry(original.id)
|
||
|
|
#expect(await eventually { room.outbox.first?.seenOnServer == true })
|
||
|
|
#expect(room.outbox.first?.createdAt == original.createdAt)
|
||
|
|
#expect(await client.submittedIDs().allSatisfy { $0 == original.requestID })
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test @MainActor func disconnectCannotRestartOldClientAndStaleReadCannotPoisonNewRun() async throws {
|
||
|
|
let old = TestAgentClient(); await old.pauseRead()
|
||
|
|
let room = SybilAgentRoom(); room.configure(client: old, storage: MemoryOutbox().storage)
|
||
|
|
for _ in 0..<100 {
|
||
|
|
if await old.hasHeldRead() { break }
|
||
|
|
try? await Task.sleep(for: .milliseconds(10))
|
||
|
|
}
|
||
|
|
room.disconnect()
|
||
|
|
room.setActive(false); room.setActive(true)
|
||
|
|
#expect(await old.connectedAfters().isEmpty)
|
||
|
|
let next = TestAgentClient(pages: ["0": agentPage([agentMessage("new", entry: "2")], cursor: "2")])
|
||
|
|
room.configure(client: next, storage: MemoryOutbox().storage)
|
||
|
|
defer { room.stop() }
|
||
|
|
#expect(await eventually { room.connection == "Connected" })
|
||
|
|
await old.releaseRead()
|
||
|
|
try? await Task.sleep(for: .milliseconds(50))
|
||
|
|
try await room.catchUp(client: next)
|
||
|
|
#expect(room.messages.map(\.id) == ["new"])
|
||
|
|
#expect(await next.requestedCursors() == ["0", "2"])
|
||
|
|
#expect(await old.connectedAfters().isEmpty)
|
||
|
|
}
|
||
|
|
|
||
|
|
@Test @MainActor func socketReplayAfterForegroundDeduplicatesAndRetainsMergedCursor() async throws {
|
||
|
|
let client = TestAgentClient()
|
||
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
||
|
|
defer { room.stop() }
|
||
|
|
#expect(await eventually { room.connection == "Connected" })
|
||
|
|
let socket = try #require(await client.latestSocket())
|
||
|
|
let frame = #"{"type":"messages","conversationId":"root","messages":[{"id":"late","createdAt":"2026-10-14T18:00:00Z","role":"assistant","content":"Late reply","metadata":{"kind":"message","entryId":7}}],"cursor":"7","beforeCursor":null,"hasMore":false}"#
|
||
|
|
await socket.push(frame)
|
||
|
|
#expect(await eventually { room.snapshot.room.cursor == "7" })
|
||
|
|
room.setActive(false); room.setActive(true)
|
||
|
|
#expect(await eventually { room.connection == "Connected" })
|
||
|
|
let next = try #require(await client.latestSocket())
|
||
|
|
await next.push(frame)
|
||
|
|
try? await Task.sleep(for: .milliseconds(150))
|
||
|
|
#expect(await client.connectedAfters() == ["0", "7"])
|
||
|
|
#expect(room.messages.count == 1)
|
||
|
|
#expect(await socket.closed)
|
||
|
|
}
|
||
|
|
|
||
|
|
}
|