Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
66 changes: 62 additions & 4 deletions Sources/Feed/FeedCoordinator.swift
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,10 @@ final class FeedCoordinator: @unchecked Sendable {
/// Main-actor isolated: read/written only from the `@MainActor` attention
/// methods.
@MainActor private var pendingAttentionStates: [FeedAttentionTarget: AttentionOverlayState] = [:]
/// Codex owns its TUI approval prompt, so its zero-wait PermissionRequest
/// telemetry has no Feed waiter to clear the needs-input overlay. Keep one
/// transient target per agent session and retire it on the next event.
@MainActor private var transientAttentionTargets: [String: FeedAttentionTarget] = [:]

/// Tail of the serialized `CMUXFeedQuestion.` category mutation chain.
/// `UNUserNotificationCenter` has no atomic category merge, so every
Expand Down Expand Up @@ -163,12 +167,16 @@ final class FeedCoordinator: @unchecked Sendable {

@MainActor
private func acceptOnMainActor(
_ event: WorkstreamEvent
_ event: WorkstreamEvent,
transientAttentionTarget: (ownerId: UUID, surfaceId: UUID?)? = nil
) -> FeedEventAcceptance {
switch resolveDeliveryTarget(for: [event]) {
case .accepted(let events):
guard let revalidatedEvent = events.first,
let item = ingestRevalidatedOnMainActor(revalidatedEvent) else {
let item = ingestRevalidatedOnMainActor(
revalidatedEvent,
transientAttentionTarget: transientAttentionTarget
) else {
return .unavailable
}
return .accepted(event: revalidatedEvent, item: item)
Expand All @@ -181,14 +189,23 @@ final class FeedCoordinator: @unchecked Sendable {

/// Inserts a revalidated event and returns the item the store now holds for it.
@MainActor
func ingestRevalidatedOnMainActor(_ event: WorkstreamEvent) -> WorkstreamItem? {
func ingestRevalidatedOnMainActor(
_ event: WorkstreamEvent,
transientAttentionTarget: (ownerId: UUID, surfaceId: UUID?)? = nil
) -> WorkstreamItem? {
guard let store else { return nil }
guard let item = store.ingestReturningItem(event) else { return nil }
retireTransientAttention(for: event)
observeSemanticLifecycle(event)
let retiredDecision = retirePendingDecisionsSuperseded(by: event)
if !retiredDecision {
clearAgentPromptNotificationsSuperseded(by: event)
}
if let transientAttentionTarget,
Self.shouldSurfaceTransientAttention(for: event),
let target = surfaceTransientAttention(event: event, resolved: transientAttentionTarget) {
transientAttentionTargets[Self.transientAttentionKey(for: event)] = target
}
if let ppid = event.ppid, ppid > 0 {
armPidWatcher(ppid: ppid)
}
Expand Down Expand Up @@ -410,6 +427,9 @@ final class FeedCoordinator: @unchecked Sendable {
onAcceptedOnMainActor: @escaping @MainActor @Sendable (WorkstreamEvent) -> Void,
onAccepted: @escaping @Sendable (WorkstreamEvent) -> Void
) -> Bool {
let transientAttentionTarget = Self.shouldSurfaceTransientAttention(for: event)
? Self.resolveAttentionTargetSynchronously(event: event)
: nil
return feedIngressDeliveryLane.enqueueZeroWait(
metadata: Self.ingressMetadata(
for: [event],
Expand All @@ -419,7 +439,10 @@ final class FeedCoordinator: @unchecked Sendable {
let acceptedEvent: WorkstreamEvent? = DispatchQueue.main.sync {
MainActor.assumeIsolated {
let accept: () -> WorkstreamEvent? = {
guard case .accepted(let event, _) = FeedCoordinator.shared.acceptOnMainActor(event) else {
guard case .accepted(let event, _) = FeedCoordinator.shared.acceptOnMainActor(
event,
transientAttentionTarget: transientAttentionTarget
) else {
return nil
}
return event
Expand Down Expand Up @@ -666,6 +689,19 @@ extension FeedCoordinator {
}
}

/// Codex keeps its approval prompt in the terminal, so its PermissionRequest
/// hook is zero-wait telemetry rather than a Feed waiter. Surface that one
/// provider's decision as transient attention until the session emits its
/// next lifecycle event. Other providers keep their existing blocking path.
static func shouldSurfaceTransientAttention(for event: WorkstreamEvent) -> Bool {
event.source.caseInsensitiveCompare("codex") == .orderedSame
&& isBlockingDecisionEvent(event.hookEventName)
}

private static func transientAttentionKey(for event: WorkstreamEvent) -> String {
"\(event.source)\u{0}\(event.sessionId)"
}

/// Maps a feed `source` (agent id) to the agent-lifecycle status key the
/// sidebar reads. Claude reports under `claude_code`; every other agent
/// keys its status by its own source name. Returning the agent's own key
Expand Down Expand Up @@ -825,6 +861,28 @@ extension FeedCoordinator {
return target
}

@MainActor
private func surfaceTransientAttention(
event: WorkstreamEvent,
resolved: (ownerId: UUID, surfaceId: UUID?)
) -> FeedAttentionTarget? {
let tabManager = AppDelegate.shared?.tabManagerFor(tabId: resolved.ownerId)
?? AppDelegate.shared?.tabManagerFor(windowId: resolved.ownerId)
return surfaceBlockingDecisionAttention(
event: event,
resolved: resolved,
tabManager: tabManager
)
}

@MainActor
private func retireTransientAttention(for event: WorkstreamEvent) {
guard let target = transientAttentionTargets.removeValue(
forKey: Self.transientAttentionKey(for: event)
) else { return }
concludeBlockingDecisionAttention(target)
}

/// Concludes a blocking decision's attention overlay. Decrements the
/// per-target refcount and, when it reaches zero, clears the needs-input
/// overlay. Feed owns a reserved lifecycle/status slot, so cleanup removes
Expand Down
57 changes: 57 additions & 0 deletions cmuxTests/FeedCoordinatorTests.swift
Original file line number Diff line number Diff line change
Expand Up @@ -640,6 +640,41 @@ struct FeedCoordinatorTests {
#expect(inserted)
}

@Test func zeroWaitCodexPermissionSurfacesTransientNeedsInputAttention() async {
defer { Self.resetFeedCoordinatorTestHooks() }
let attention = AttentionSurfaceRecorder()

await MainActor.run {
FeedCoordinator.shared.install(store: WorkstreamStore(ringCapacity: 10))
FeedCoordinatorTestHooks.attentionSurfaceObserver = { event in
attention.record(event)
}
}

let event = WorkstreamEvent(
sessionId: "codex-zero-wait-permission",
hookEventName: .permissionRequest,
source: "codex",
workspaceId: UUID().uuidString,
surfaceId: UUID().uuidString,
toolName: "Bash",
toolInputJSON: #"{"command":"touch /tmp/cmux-codex"}"#,
requestId: "codex-zero-wait-permission-request"
)
let done = DispatchSemaphore(value: 0)
DispatchQueue.global(qos: .userInitiated).async {
_ = FeedCoordinator.shared.ingestBlocking(event: event, waitTimeout: 0)
done.signal()
}

#expect(done.wait(timeout: .now() + 2) == .success)
#expect(
attention.events.count == 1,
"Codex TUI approvals must surface needs-input attention even without a Feed waiter"
)
#expect(attention.events.first?.source == "codex")
}

@Test func zeroWaitOneWayTelemetryDoesNotSynchronouslyHopToMainActor() async {
await MainActor.run {
let store = WorkstreamStore(ringCapacity: 10)
Expand Down Expand Up @@ -906,6 +941,28 @@ struct FeedCoordinatorTests {
#expect(!FeedCoordinator.isBlockingDecisionEvent(.userPromptSubmit))
}

@Test func transientAttentionIsLimitedToCodexDecisionTelemetry() {
let codex = WorkstreamEvent(
sessionId: "codex-transient-classifier",
hookEventName: .permissionRequest,
source: "codex"
)
let claude = WorkstreamEvent(
sessionId: "claude-blocking-classifier",
hookEventName: .permissionRequest,
source: "claude"
)
let tool = WorkstreamEvent(
sessionId: "codex-tool-classifier",
hookEventName: .preToolUse,
source: "codex"
)

#expect(FeedCoordinator.shouldSurfaceTransientAttention(for: codex))
#expect(!FeedCoordinator.shouldSurfaceTransientAttention(for: claude))
#expect(!FeedCoordinator.shouldSurfaceTransientAttention(for: tool))
}

@Test func lifecycleStatusKeyMatchesAgentReportedKey() {
// Claude reports its lifecycle under `claude_code`; normalize the Feed
// source the same way without mutating that agent-owned slot.
Expand Down
Loading