Skip to content

Commit 5c99169

Browse files
steipetesteipete
authored
feat(gateway): watched sessions in presence with viewer facepiles (#111225)
* feat(gateway): expose watched sessions in presence * fix(gateway): resolve concurrent watched-session rollbacks via in-flight state * fix(ui): viewing CI conformance — lazy facepile, bindings, dead exports * fix(ui): move viewer presence projection out of the startup chunk * fix(ui): capture sessionKey before facepile filter closure
1 parent cfe4096 commit 5c99169

21 files changed

Lines changed: 797 additions & 88 deletions

File tree

‎apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -964,6 +964,7 @@ public struct PresenceEntry: Codable, Sendable {
964964
public let scopes: [String]?
965965
public let instanceid: String?
966966
public let user: [String: AnyCodable]?
967+
public let watchedsessions: [String]?
967968

968969
public init(
969970
host: String? = nil,
@@ -982,7 +983,8 @@ public struct PresenceEntry: Codable, Sendable {
982983
roles: [String]? = nil,
983984
scopes: [String]? = nil,
984985
instanceid: String? = nil,
985-
user: [String: AnyCodable]? = nil)
986+
user: [String: AnyCodable]? = nil,
987+
watchedsessions: [String]? = nil)
986988
{
987989
self.host = host
988990
self.ip = ip
@@ -1001,6 +1003,7 @@ public struct PresenceEntry: Codable, Sendable {
10011003
self.scopes = scopes
10021004
self.instanceid = instanceid
10031005
self.user = user
1006+
self.watchedsessions = watchedsessions
10041007
}
10051008

10061009
private enum CodingKeys: String, CodingKey {
@@ -1021,6 +1024,7 @@ public struct PresenceEntry: Codable, Sendable {
10211024
case scopes
10221025
case instanceid = "instanceId"
10231026
case user
1027+
case watchedsessions = "watchedSessions"
10241028
}
10251029
}
10261030

‎packages/gateway-protocol/src/schema/snapshot.test.ts‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,4 +28,16 @@ describe("SnapshotSchema", () => {
2828
it("keeps presence user identity optional", () => {
2929
expect(Value.Check(SnapshotSchema, snapshotWithPresence({ ts: 1 }))).toBe(true);
3030
});
31+
32+
it("accepts optional watched session keys", () => {
33+
expect(
34+
Value.Check(
35+
SnapshotSchema,
36+
snapshotWithPresence({
37+
ts: 1,
38+
watchedSessions: ["agent:main:main", "agent:main:work"],
39+
}),
40+
),
41+
).toBe(true);
42+
});
3143
});

‎packages/gateway-protocol/src/schema/snapshot.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,8 @@ export const PresenceEntrySchema = closedObject({
3737
avatarUrl: Type.Optional(NonEmptyString),
3838
}),
3939
),
40+
/** Session keys this connection is actively subscribed to (watching). Sorted lexicographically for deterministic snapshots. */
41+
watchedSessions: Type.Optional(Type.Array(NonEmptyString)),
4042
});
4143

4244
/** Health snapshot is intentionally opaque because providers contribute nested shapes. */

‎src/gateway/server-chat-state.test.ts‎

Lines changed: 65 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,7 @@ describe("createSessionMessageSubscriberRegistry", () => {
3131
subscribers.subscribe("conn-other", "agent:main:child", { includeApprovals: true });
3232

3333
subscribers.unsubscribeAll("conn-reviewer");
34+
expect([...subscribers.getForConnection("conn-reviewer")]).toEqual([]);
3435
expect([...subscribers.get("agent:main:main")]).toEqual([]);
3536
expect([...subscribers.getApprovals("agent:main:main")]).toEqual([]);
3637
expect([...subscribers.get("agent:main:child")]).toEqual(["conn-other"]);
@@ -41,27 +42,73 @@ describe("createSessionMessageSubscriberRegistry", () => {
4142
expect([...subscribers.getApprovals("agent:main:child")]).toEqual([]);
4243
});
4344

44-
it("rolls a provisional subscription back to its exact prior state", () => {
45+
it.each(["first", "second"])(
46+
"removes a first-time subscription when both concurrent replays fail (%s rollback first)",
47+
(firstRollback) => {
48+
const subscribers = createSessionMessageSubscriberRegistry();
49+
const first = subscribers.subscribe("conn", "agent:main:main", { provisional: true })!;
50+
const second = subscribers.subscribe("conn", "agent:main:main", { provisional: true })!;
51+
52+
if (firstRollback === "first") {
53+
first();
54+
second();
55+
} else {
56+
second();
57+
first();
58+
}
59+
60+
expect([...subscribers.get("agent:main:main")]).toEqual([]);
61+
expect([...subscribers.getForConnection("conn")]).toEqual([]);
62+
},
63+
);
64+
65+
it.each(["first", "second"])(
66+
"keeps the successful concurrent replay recency (%s resolution first)",
67+
(firstResolution) => {
68+
const subscribers = createSessionMessageSubscriberRegistry();
69+
subscribers.subscribe("conn", "agent:main:other");
70+
const first = subscribers.subscribe("conn", "agent:main:main", { provisional: true })!;
71+
const second = subscribers.subscribe("conn", "agent:main:main", { provisional: true })!;
72+
73+
if (firstResolution === "first") {
74+
first();
75+
second.commit();
76+
} else {
77+
second.commit();
78+
first();
79+
}
80+
81+
expect([...subscribers.getForConnection("conn")]).toEqual([
82+
"agent:main:other",
83+
"agent:main:main",
84+
]);
85+
},
86+
);
87+
88+
it("retains the committed recency when a re-subscribe replay fails", () => {
4589
const subscribers = createSessionMessageSubscriberRegistry();
90+
subscribers.subscribe("conn", "agent:main:main");
91+
subscribers.subscribe("conn", "agent:main:child");
92+
const rollback = subscribers.subscribe("conn", "agent:main:main", { provisional: true })!;
4693

47-
const removeNew = subscribers.subscribe("conn-new", "agent:main:main", {
48-
includeApprovals: true,
49-
});
50-
removeNew?.();
51-
expect([...subscribers.get("agent:main:main")]).toEqual([]);
52-
expect([...subscribers.getApprovals("agent:main:main")]).toEqual([]);
94+
rollback();
5395

54-
subscribers.subscribe("conn-plain", "agent:main:main");
55-
const restorePlain = subscribers.subscribe("conn-plain", "agent:main:main", {
56-
includeApprovals: true,
57-
});
58-
restorePlain?.();
59-
expect([...subscribers.get("agent:main:main")]).toEqual(["conn-plain"]);
60-
expect([...subscribers.getApprovals("agent:main:main")]).toEqual([]);
96+
expect([...subscribers.getForConnection("conn")]).toEqual([
97+
"agent:main:main",
98+
"agent:main:child",
99+
]);
100+
});
61101

62-
subscribers.subscribe("conn-reviewer", "agent:main:main", { includeApprovals: true });
63-
const restoreReviewer = subscribers.subscribe("conn-reviewer", "agent:main:main");
64-
restoreReviewer?.();
65-
expect([...subscribers.getApprovals("agent:main:main")]).toEqual(["conn-reviewer"]);
102+
it("does not restore a replay invalidated by unsubscribe", () => {
103+
const subscribers = createSessionMessageSubscriberRegistry();
104+
const subscription = subscribers.subscribe("conn", "agent:main:main", {
105+
provisional: true,
106+
})!;
107+
108+
subscribers.unsubscribe("conn", "agent:main:main");
109+
subscription.commit();
110+
111+
expect([...subscribers.getForConnection("conn")]).toEqual([]);
112+
expect([...subscribers.get("agent:main:main")]).toEqual([]);
66113
});
67114
});

0 commit comments

Comments
 (0)