2026-10-14 13:34:37 -07:00
|
|
|
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"
|
2026-10-14 22:03:21 -07:00
|
|
|
var readSource: String?
|
|
|
|
|
var readWrites: [(cursor: String, source: String?)] = []
|
|
|
|
|
var writeReply: AgentReadState?
|
|
|
|
|
var writeConflict: Bool?
|
|
|
|
|
var holdWrite = false
|
|
|
|
|
var heldWrite: CheckedContinuation<Void, Never>?
|
2026-10-14 13:34:37 -07:00
|
|
|
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 } }
|
2026-10-14 22:03:21 -07:00
|
|
|
return AgentReadState(cursor: read, sourceId: readSource)
|
2026-10-14 13:34:37 -07:00
|
|
|
}
|
|
|
|
|
func pauseRead() { holdRead = true }
|
2026-10-14 22:03:21 -07:00
|
|
|
func releaseRead() { holdRead = false; heldRead?.resume(returning: AgentReadState(cursor: read, sourceId: readSource)); heldRead = nil }
|
2026-10-14 13:34:37 -07:00
|
|
|
func hasHeldRead() -> Bool { heldRead != nil }
|
2026-10-14 22:03:21 -07:00
|
|
|
func agentMarkRead(cursor: String) async throws -> AgentReadState {
|
|
|
|
|
try await agentMarkRead(cursor: cursor, sourceID: nil)
|
2026-10-14 13:34:37 -07:00
|
|
|
}
|
2026-10-14 22:03:21 -07:00
|
|
|
func agentMarkRead(cursor: String, sourceID: String?) async throws -> AgentReadState {
|
|
|
|
|
readWrites.append((cursor, sourceID))
|
|
|
|
|
if holdWrite { await withCheckedContinuation { heldWrite = $0 } }
|
|
|
|
|
if let coded = writeConflict {
|
|
|
|
|
if coded { throw AgentAttachmentIssue(message: "Source mismatch", code: "source_mismatch", statusCode: 409) }
|
|
|
|
|
throw APIError.httpError(statusCode: 409, message: "Source mismatch")
|
|
|
|
|
}
|
|
|
|
|
if failRead { throw AgentFailure(message: "Offline") }
|
|
|
|
|
if let writeReply { return writeReply }
|
|
|
|
|
read = try AgentCursor.advance(read, cursor)
|
|
|
|
|
return AgentReadState(cursor: read, sourceId: readSource)
|
|
|
|
|
}
|
|
|
|
|
func setReadState(_ state: AgentReadState) { read = state.cursor; readSource = state.sourceId }
|
|
|
|
|
func setWriteReply(_ state: AgentReadState?) { writeReply = state }
|
|
|
|
|
func setWriteConflict(coded: Bool) { writeConflict = coded }
|
|
|
|
|
func pauseWrite() { holdWrite = true }
|
|
|
|
|
func releaseWrite() { holdWrite = false; heldWrite?.resume(); heldWrite = nil }
|
|
|
|
|
func hasHeldWrite() -> Bool { heldWrite != nil }
|
|
|
|
|
func writtenReads() -> [(cursor: String, source: String?)] { readWrites }
|
|
|
|
|
func setHistory(_ pages: [String: AgentPage]) { self.pages = pages }
|
2026-10-14 13:34:37 -07:00
|
|
|
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 {
|
2026-10-14 15:31:24 -07:00
|
|
|
@Test @MainActor func mixedLoadedHistoryFiltersOnlyToolRows() async throws {
|
|
|
|
|
var user = agentMessage("user", entry: "1", role: .user)
|
|
|
|
|
user.content = "Explain the tool_call and tool result."
|
|
|
|
|
let call = agentMessage("call", entry: "2", role: .tool, info: ["status": .string("initiated")])
|
|
|
|
|
var reply = agentMessage("reply", entry: "3")
|
|
|
|
|
reply.content = "The search tool failed; here is what you can do."
|
|
|
|
|
// Role-only results still need hiding even without tool-call metadata.
|
|
|
|
|
var result = agentMessage("result", entry: "4", role: .tool)
|
|
|
|
|
result.metadata = nil
|
|
|
|
|
let markedCall = agentMessage("marked-call", entry: "5", info: ["kind": .string("tool_call")])
|
|
|
|
|
var plainReply = agentMessage("plain-reply", entry: "6")
|
|
|
|
|
plainReply.metadata = nil
|
|
|
|
|
let failure = agentMessage("failure", entry: "7", info: ["status": .string("failed"), "error": .string("Please try again")])
|
|
|
|
|
let terminal = agentMessage("terminal", entry: "8", role: .tool, info: ["status": .string("failed"), "error": .string("Raw tool failure"), "resultPreview": .string("Raw result")])
|
|
|
|
|
let history = [user, call, reply, result, markedCall, plainReply, failure, terminal].enumerated().map { index, message in
|
|
|
|
|
var message = message
|
|
|
|
|
message.createdAt.addTimeInterval(Double(index))
|
|
|
|
|
return message
|
|
|
|
|
}
|
|
|
|
|
let client = TestAgentClient(pages: ["0": agentPage(Array(history.dropLast()), cursor: "7", more: true), "7": agentPage([history[7]], cursor: "8")])
|
|
|
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
|
|
|
|
defer { room.stop() }
|
|
|
|
|
#expect(await eventually { room.connection == "Connected" })
|
|
|
|
|
#expect(await client.requestedCursors() == ["0", "7"])
|
|
|
|
|
#expect(await client.connectedAfters() == ["8"])
|
|
|
|
|
let transcript = AgentTranscript(messages: room.messages)
|
|
|
|
|
#expect(transcript.displayed == [history[0], history[2], history[5], history[6]])
|
|
|
|
|
#expect(room.snapshot.room.messages == history)
|
|
|
|
|
#expect(room.messages.count == 8)
|
|
|
|
|
#expect(room.snapshot.room.cursor == "8")
|
|
|
|
|
#expect(room.snapshot.room.readCursor == "0")
|
|
|
|
|
#expect(room.unreadCount == 2)
|
|
|
|
|
#expect(room.snapshot.entranceIDs.isEmpty)
|
|
|
|
|
// Shared regular-chat tool metadata remains available and unchanged.
|
|
|
|
|
#expect(terminal.isToolCallLog)
|
|
|
|
|
#expect(terminal.toolCallMetadata?.resultPreview == "Raw result")
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Test func transcriptWindowsAndAnchorsUseVisibleRows() {
|
|
|
|
|
var raw: [Message] = [], visible: [Message] = []
|
|
|
|
|
for index in 1...160 {
|
|
|
|
|
var message = agentMessage("reply-\(index)", entry: "\(index * 2)")
|
|
|
|
|
message.createdAt.addTimeInterval(Double(index * 2) * 86_400)
|
|
|
|
|
var tool = agentMessage("tool-\(index)", entry: "\(index * 2 + 1)", role: .tool)
|
|
|
|
|
// Tools on separate dates must not introduce their own separators.
|
|
|
|
|
tool.createdAt = message.createdAt.addingTimeInterval(86_400)
|
|
|
|
|
raw += [message, tool]; visible.append(message)
|
|
|
|
|
}
|
|
|
|
|
let initial = AgentTranscript(messages: raw)
|
|
|
|
|
#expect(initial.startIndex == 60)
|
|
|
|
|
#expect(initial.displayed == Array(visible.suffix(100)))
|
|
|
|
|
let anchorID = initial.displayed.first?.id
|
|
|
|
|
let hidden = agentMessage("late-tool", entry: "400", role: .tool)
|
|
|
|
|
let updated = AgentTranscript(messages: raw + [hidden], startID: anchorID)
|
|
|
|
|
#expect(updated.displayed == initial.displayed)
|
|
|
|
|
let earlierID = initial.messages[max(0, initial.startIndex - 100)].id
|
|
|
|
|
let expanded = AgentTranscript(messages: raw, startID: earlierID, limit: 200)
|
|
|
|
|
#expect(expanded.displayed == visible)
|
|
|
|
|
#expect(expanded.displayed.firstIndex(where: { $0.id == anchorID }) == 60)
|
|
|
|
|
let hiddenAnchor = AgentTranscript(messages: raw, startID: "tool-160")
|
|
|
|
|
#expect(hiddenAnchor.displayed == initial.displayed)
|
|
|
|
|
let onlyTools = AgentTranscript(messages: raw.filter { $0.role == .tool })
|
|
|
|
|
#expect(onlyTools.messages.isEmpty)
|
|
|
|
|
#expect(onlyTools.displayed.isEmpty)
|
|
|
|
|
#expect(onlyTools.startIndex == 0)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Test @MainActor func liveToolUpsertsPreserveRawHistoryCursorsAndActivity() async throws {
|
|
|
|
|
let user = agentMessage("user", entry: "1", role: .user)
|
|
|
|
|
let reply = agentMessage("reply", entry: "2")
|
|
|
|
|
let client = TestAgentClient(pages: ["0": agentPage([user, reply], cursor: "2")])
|
|
|
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
|
|
|
|
defer { room.stop() }
|
|
|
|
|
#expect(await eventually { room.connection == "Connected" })
|
|
|
|
|
let queued = agentMessage("queued", entry: "0", role: .user, info: ["status": .string("queued")])
|
|
|
|
|
var activity = AgentActivity()
|
|
|
|
|
activity.isThinking = true; activity.backgroundWorkers = 2; activity.queued = 1
|
|
|
|
|
activity.queuedMessages = [queued]
|
|
|
|
|
activity.wait = .init(due: 1_800_000_000_000, reason: "Waiting for background work")
|
|
|
|
|
activity.error = "A worker needs attention"
|
|
|
|
|
try room.receive(.activity("root", activity)); room.flush()
|
|
|
|
|
#expect(await eventually { room.snapshot.isTyping })
|
|
|
|
|
let visible = AgentTranscript(messages: room.messages).displayed
|
|
|
|
|
#expect(Set(visible.map(\.id)) == ["user", "reply", "queued"])
|
|
|
|
|
let start = agentMessage("tool", entry: "3", role: .tool, info: ["status": .string("initiated"), "toolName": .string("search"), "args": .object(["query": .string("fixture")]), "startedAt": .string("start")])
|
|
|
|
|
let completed = agentMessage("tool", entry: "4", role: .tool, info: ["status": .string("completed"), "resultPreview": .string("Raw preview")])
|
|
|
|
|
let result = agentMessage("tool", entry: "5", role: .tool, info: ["status": .string("completed"), "result": .object(["matches": .number(3)])])
|
|
|
|
|
for (message, cursor) in [(start, "3"), (completed, "4"), (result, "5"), (start, "3")] {
|
|
|
|
|
try room.receive(.messages(agentPage([message], cursor: cursor))); room.flush()
|
|
|
|
|
#expect(AgentTranscript(messages: room.messages).displayed == visible)
|
|
|
|
|
#expect(room.snapshot.room.messages.count == 3)
|
|
|
|
|
#expect(room.unreadCount == 1)
|
|
|
|
|
#expect(room.snapshot.room.readCursor == "0")
|
|
|
|
|
#expect(room.snapshot.isTyping)
|
|
|
|
|
#expect(room.snapshot.room.activity == activity)
|
|
|
|
|
}
|
|
|
|
|
let tool = try #require(room.snapshot.room.messages.first(where: { $0.id == "tool" })?.toolCallMetadata)
|
|
|
|
|
#expect(tool.status == "completed")
|
|
|
|
|
#expect(tool.args == .object(["query": .string("fixture")]))
|
|
|
|
|
#expect(tool.resultPreview == "Raw preview")
|
|
|
|
|
#expect(tool.result == .object(["matches": .number(3)]))
|
|
|
|
|
#expect(tool.startedAt == "start")
|
|
|
|
|
#expect(room.snapshot.room.cursor == "5")
|
|
|
|
|
// The rendered snapshot's cursor includes hidden tool updates. A later
|
|
|
|
|
// reply must remain unread when only that captured cursor is acknowledged.
|
|
|
|
|
room.isSelected = true
|
|
|
|
|
let capturedCursor = room.snapshot.room.cursor
|
|
|
|
|
let lateReply = agentMessage("late-reply", entry: "6")
|
|
|
|
|
let lateTool = agentMessage("late-tool", entry: "7", role: .tool)
|
|
|
|
|
try room.receive(.messages(agentPage([lateReply, lateTool], cursor: "7"))); room.flush()
|
|
|
|
|
room.markRead(renderedCursor: capturedCursor)
|
|
|
|
|
#expect(await eventually { room.snapshot.room.readCursor == "5" })
|
|
|
|
|
#expect(room.snapshot.room.cursor == "7")
|
|
|
|
|
#expect(room.unreadCount == 1)
|
|
|
|
|
#expect(AgentTranscript(messages: room.messages).displayed == visible + [lateReply])
|
|
|
|
|
#expect(!room.snapshot.isTyping)
|
|
|
|
|
#expect(room.isBusy)
|
|
|
|
|
let socket = try #require(await client.latestSocket())
|
|
|
|
|
await socket.push(#"{"type":"error","error":"Room connection unavailable"}"#)
|
|
|
|
|
#expect(await eventually { room.error == "Room connection unavailable" })
|
|
|
|
|
#expect(AgentTranscript(messages: room.messages).displayed == visible + [lateReply])
|
|
|
|
|
#expect(room.snapshot.room.messages.count == 5)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Test @MainActor func failedSendDetailsAndRetryStayVisibleWithHiddenTools() async throws {
|
|
|
|
|
let tool = agentMessage("tool", entry: "1", role: .tool)
|
|
|
|
|
let client = TestAgentClient(pages: ["0": agentPage([tool], cursor: "1")])
|
|
|
|
|
await client.setFailures(submit: true)
|
|
|
|
|
let memory = MemoryOutbox()
|
|
|
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: memory.storage)
|
|
|
|
|
defer { room.stop() }
|
|
|
|
|
#expect(await eventually { room.connection == "Connected" })
|
|
|
|
|
#expect(room.send("Please retry the search tool"))
|
|
|
|
|
#expect(await eventually { room.outbox.first?.status == "failed" })
|
|
|
|
|
let failed = try #require(AgentTranscript(messages: room.messages).displayed.first)
|
|
|
|
|
let requestID = try #require(room.outbox.first?.requestID)
|
|
|
|
|
#expect(failed.role == .user)
|
|
|
|
|
#expect(failed.content == "Please retry the search tool")
|
|
|
|
|
#expect(failed.metadata?.objectValue?["status"] == .string("failed"))
|
|
|
|
|
#expect(failed.metadata?.objectValue?["error"] == .string("Lost acknowledgement"))
|
|
|
|
|
#expect(AgentTranscript(messages: room.messages).displayed.count == 1)
|
|
|
|
|
#expect(memory.read().first?.requestID == requestID)
|
|
|
|
|
await client.setFailures()
|
|
|
|
|
room.retry(failed.id)
|
|
|
|
|
#expect(await eventually { room.outbox.first?.status == "queued" })
|
|
|
|
|
let retried = try #require(AgentTranscript(messages: room.messages).displayed.first)
|
|
|
|
|
#expect(retried.id == failed.id)
|
|
|
|
|
#expect(retried.createdAt == failed.createdAt)
|
|
|
|
|
#expect(retried.metadata?.objectValue?["status"] == .string("queued"))
|
|
|
|
|
#expect(retried.metadata?.objectValue?["error"] == nil)
|
|
|
|
|
#expect(await client.submittedIDs() == [requestID, requestID])
|
|
|
|
|
#expect(room.snapshot.room.messages == [tool])
|
|
|
|
|
#expect(room.messages.count == 2)
|
|
|
|
|
}
|
|
|
|
|
|
2026-10-14 13:34:37 -07:00
|
|
|
@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)
|
|
|
|
|
}
|
2026-10-14 22:03:21 -07:00
|
|
|
|
|
|
|
|
@Test @MainActor func legacyReadPromotionCannotRelabelQueuedHistoryAndRequiresFreshReload() async throws {
|
|
|
|
|
let sourceB = "22222222-2222-4222-8222-222222222222"
|
|
|
|
|
let client = TestAgentClient(pages: ["0": agentPage([agentMessage("old", entry: "100")], cursor: "100")])
|
|
|
|
|
await client.setReadState(.init(cursor: "90"))
|
|
|
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
|
|
|
|
defer { room.stop() }
|
|
|
|
|
#expect(await eventually { room.connection == "Connected" })
|
|
|
|
|
room.isSelected = true; await client.pauseWrite(); room.markRead(renderedCursor: "100")
|
|
|
|
|
for _ in 0..<100 {
|
|
|
|
|
if await client.hasHeldWrite() { break }
|
|
|
|
|
try? await Task.sleep(for: .milliseconds(10))
|
|
|
|
|
}
|
|
|
|
|
#expect(await client.hasHeldWrite())
|
|
|
|
|
#expect(throws: AgentFailure.self) { try room.applyRead(.init(cursor: "0", sourceId: sourceB)) }
|
|
|
|
|
#expect(room.requiresReload && room.connection == "Reload required")
|
|
|
|
|
#expect(room.snapshot.room.readCursor == "90" && room.snapshot.room.cursor == "100")
|
|
|
|
|
await client.setWriteReply(.init(cursor: "100", sourceId: sourceB)); await client.releaseWrite()
|
|
|
|
|
room.markRead(renderedCursor: "100"); room.setActive(false); room.setActive(true)
|
|
|
|
|
try await Task.sleep(for: .milliseconds(50))
|
|
|
|
|
#expect(room.requiresReload && room.snapshot.room.readCursor == "90")
|
|
|
|
|
#expect(await client.writtenReads().count == 1)
|
|
|
|
|
#expect(await client.writtenReads().first?.source == nil)
|
|
|
|
|
#expect(room.messages.map(\.id) == ["old"])
|
|
|
|
|
|
|
|
|
|
// Explicit reconnect resets cursor space before discovering B or loading
|
|
|
|
|
// its history. The old queued100 can never be sent under sourceB.
|
|
|
|
|
await client.setWriteReply(nil); await client.setReadState(.init(cursor: "0", sourceId: sourceB))
|
|
|
|
|
await client.setHistory(["0": agentPage([agentMessage("new", entry: "1")], cursor: "1")])
|
|
|
|
|
room.retryConnection()
|
|
|
|
|
#expect(await eventually { room.connection == "Connected" })
|
|
|
|
|
#expect(!room.requiresReload && room.snapshot.room.readCursor == "0")
|
|
|
|
|
#expect(room.messages.map(\.id) == ["new"])
|
|
|
|
|
room.markRead(renderedCursor: "1")
|
|
|
|
|
#expect(await eventually { room.snapshot.room.readCursor == "1" })
|
|
|
|
|
let writes = await client.writtenReads()
|
|
|
|
|
#expect(writes.count == 2 && writes.last?.cursor == "1" && writes.last?.source == sourceB)
|
|
|
|
|
#expect(!writes.contains { $0.cursor == "100" && $0.source == sourceB })
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Test @MainActor func historyBeforeReadPinsLegacySourceBeforePublishing() throws {
|
|
|
|
|
let room = SybilAgentRoom(); defer { room.stop() }
|
|
|
|
|
try room.receive(.messages(agentPage([agentMessage("legacy", entry: "100")], cursor: "100")))
|
|
|
|
|
#expect(throws: AgentFailure.self) {
|
|
|
|
|
try room.applyRead(.init(cursor: "0", sourceId: "22222222-2222-4222-8222-222222222222"))
|
|
|
|
|
}
|
|
|
|
|
#expect(room.requiresReload && room.snapshot.room.cursor == "100" && room.snapshot.room.readCursor == "0")
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Test @MainActor func mismatchedAndNonAdvancingReadAcknowledgmentsStopWriter() async throws {
|
|
|
|
|
let sourceA = "11111111-1111-4111-8111-111111111111"
|
|
|
|
|
let sourceB = "22222222-2222-4222-8222-222222222222"
|
|
|
|
|
for reply in [AgentReadState(cursor: "100", sourceId: sourceB), .init(cursor: "90", sourceId: sourceA), .init(cursor: "100"), .init(cursor: "invalid", sourceId: sourceA)] {
|
|
|
|
|
let client = TestAgentClient(pages: ["0": agentPage([agentMessage("old", entry: "100")], cursor: "100")])
|
|
|
|
|
await client.setReadState(.init(cursor: "90", sourceId: sourceA)); await client.setWriteReply(reply)
|
|
|
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
|
|
|
|
#expect(await eventually { room.connection == "Connected" })
|
|
|
|
|
room.isSelected = true; room.markRead(renderedCursor: "100")
|
|
|
|
|
#expect(await eventually { room.requiresReload })
|
|
|
|
|
room.markRead(renderedCursor: "100"); room.setActive(false); room.setActive(true)
|
|
|
|
|
try await Task.sleep(for: .milliseconds(30))
|
|
|
|
|
#expect(room.snapshot.room.readCursor == "90")
|
|
|
|
|
#expect(await client.writtenReads().count == 1)
|
|
|
|
|
#expect(await client.writtenReads().first?.source == sourceA)
|
|
|
|
|
room.stop()
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
@Test @MainActor func readSourceConflict409StopsWriterUntilReload() async {
|
|
|
|
|
for coded in [false, true] {
|
|
|
|
|
let client = TestAgentClient(pages: ["0": agentPage([agentMessage("old", entry: "100")], cursor: "100")])
|
|
|
|
|
await client.setReadState(.init(cursor: "90", sourceId: "11111111-1111-4111-8111-111111111111"))
|
|
|
|
|
await client.setWriteConflict(coded: coded)
|
|
|
|
|
let room = SybilAgentRoom(); room.configure(client: client, storage: MemoryOutbox().storage)
|
|
|
|
|
#expect(await eventually { room.connection == "Connected" })
|
|
|
|
|
room.isSelected = true; room.markRead(renderedCursor: "100")
|
|
|
|
|
#expect(await eventually { room.requiresReload })
|
|
|
|
|
room.markRead(renderedCursor: "100")
|
|
|
|
|
#expect(room.snapshot.room.readCursor == "90")
|
|
|
|
|
#expect(await client.writtenReads().count == 1)
|
|
|
|
|
room.stop()
|
|
|
|
|
}
|
|
|
|
|
}
|
2026-10-14 13:34:37 -07:00
|
|
|
@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)
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
}
|
