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
14 changes: 14 additions & 0 deletions apps/mobile/src/lib/realtime.ts
Original file line number Diff line number Diff line change
Expand Up @@ -316,6 +316,20 @@ function handle(event: ServerEvent) {
case "entity.changed":
for (const key of ENTITY_KEYS[event.entity] ?? []) refreshSoon({ queryKey: key });
break;
case "devices.updated":
// A computer joined, left or can take this phone now: the list of where to go when one is away follows.
void refreshGroup();
break;
}
}

async function refreshGroup() {
const token = useSession.getState().connection?.token;
try {
const me = await api.me();
if (useSession.getState().connection?.token === token) useSession.getState().setGroup(me.group);
} catch {
/* the next connect asks again */
}
}

Expand Down
10 changes: 9 additions & 1 deletion packages/core/src/remote/devices.ts
Original file line number Diff line number Diff line change
Expand Up @@ -93,6 +93,7 @@ import {
pairWithCode,
prepareRemoteMessage,
releaseChatsNow,
refreshRunnerInfo,
connectRunner,
removeRunner,
takenFrom,
Expand Down Expand Up @@ -186,7 +187,7 @@ export function startDevices(h: LinkHandler, deliver: (event: ServerEvent) => vo
replicaHint: (id) => replicaHint(id),
});
setStandInHooks({ takeOver, changed: emitGroupSoon, online: internetReachable });
startReplicaSync({ fetchChat: (peer, id) => void fetchChatFrom(peer, id) });
startReplicaSync({ fetchChat: (peer, id) => void fetchChatFrom(peer, id), refreshInfo: refreshPeer });
setLinkEventFilter(echoToPrimary);
setScheduleGroup({ active: () => groupRole() !== null, settle: settled });
setPhoneInfo(() => (getSettings().mobile.enabled ? { instance: instanceInfo(), urls: mobileUrls() } : null));
Expand Down Expand Up @@ -324,6 +325,13 @@ function phoneOf(id: string): RunnerInfo["phone"] | null {
return parseJson<RunnerInfo["phone"] | null>(getMeta(phoneKey(id)), null);
}

/** What another computer of the group says about itself, asked again: phones and windows hear when its phone address changed. */
async function refreshPeer(id: string): Promise<void> {
const before = JSON.stringify(phoneOf(id) ?? null);
await refreshRunnerInfo(id);
if (JSON.stringify(phoneOf(id) ?? null) !== before) emitGroupSoon();
}

/**
* The computers of the group a phone can talk to (GET /api/mobile/me), this one first: a phone paired with one of them
* works with all of them, so it reaches another when one is away.
Expand Down
12 changes: 11 additions & 1 deletion packages/core/src/remote/replicaSync.ts
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
import { refreshAgentFiles, trashAgentRepo } from "../agents/service";
import { bus } from "../events/bus";
import { logger } from "../log";
import { refreshMobileAccess } from "../mobile/access";
import { computerName } from "../mobile/devices";
import { reloadSchedules } from "../scheduler/scheduler";
import { notify } from "../services/notifications";
Expand Down Expand Up @@ -47,6 +48,7 @@ let watchTimer: ReturnType<typeof setInterval> | null = null;
let pullTimer: ReturnType<typeof setInterval> | null = null;
let lastSeq = 0;
let fetchChat: ((peer: string, conversationId: string) => void) | null = null;
let infoChanged: ((peer: string) => Promise<void>) | null = null;

/** Computers this one exchanges the shared data with: its primary, or (on the primary) every secondary. */
function peers(): string[] {
Expand Down Expand Up @@ -108,6 +110,8 @@ async function pullNow(peer: string): Promise<ApplyResult | null> {
// Says how far this computer got: the other one counts what is still waiting for it from there.
await l.json("POST", "/api/link/replica/ack", { after }, { timeoutMs: 30_000 }).catch(() => undefined);
mergeResults(total, retryDeferred(peer, opts));
// What it says about itself can change between connects (phone access switched on there).
await infoChanged?.(peer).catch(() => undefined);
pulledAt.set(peer, Date.now());
if (total.applied || total.failed) log.info("took changes from another computer", { peer: r.name, applied: total.applied, kept: total.kept, deferred: total.deferred, failed: total.failed });
await afterApply(r.name, total);
Expand Down Expand Up @@ -146,6 +150,10 @@ function phonesArrived(from: string): void {
setMeta(PHONE_ON_META, "1");
try {
applyRuntimeSettings(updateSettings({ mobile: { enabled: true } } as never));
// Listening now: the other computers learn where phones reach this one (their next pull asks), and tell the phones.
void refreshMobileAccess()
.catch(() => undefined)
.then(() => bus.emit({ type: "replica.changed", seq: replicaSeq() }));
notify("info", "Your phone works with this computer too", `It was paired on ${from}. When that computer is away, the Godmode app on your phone connects to ${computerName()}.`, "/settings/phone");
} catch (err) {
log.warn("could not switch phone access on for a phone of the group", err);
Expand Down Expand Up @@ -178,8 +186,9 @@ export function pendingChanges(): number {
* `chat`: asks the computer a change came from for a chat it points at (the mirror). Watches this computer's own
* changes and tells the others (`replica.changed` goes to the linked computers only), and pulls once a minute.
*/
export function startReplicaSync(opts: { fetchChat: (peer: string, conversationId: string) => void }): void {
export function startReplicaSync(opts: { fetchChat: (peer: string, conversationId: string) => void; refreshInfo: (peer: string) => Promise<void> }): void {
fetchChat = opts.fetchChat;
infoChanged = opts.refreshInfo;
ensureReplica();
lastSeq = replicaSeq();
watchTimer ??= setInterval(() => {
Expand All @@ -205,4 +214,5 @@ export function stopReplicaSync(): void {
for (const t of hintTimers.values()) clearTimeout(t);
hintTimers.clear();
fetchChat = null;
infoChanged = null;
}
9 changes: 9 additions & 0 deletions packages/core/src/remote/runners.ts
Original file line number Diff line number Diff line change
Expand Up @@ -259,6 +259,15 @@ export function runnerInfoOf(id: string): RunnerInfo | undefined {
return infos.get(id);
}

/** Ask again what a connected runner (or another computer of the group) says about itself; returns the new answer. */
export async function refreshRunnerInfo(id: string): Promise<RunnerInfo | undefined> {
const l = links.get(id);
if (!l || l.state.state !== "online") return infos.get(id);
const info = await l.json<RunnerInfo>("GET", "/api/link/info", undefined, { timeoutMs: 15_000 });
infos.set(id, info);
return info;
}

/** A chat that moved back to `runnerId` before it was let go there: nothing to let go anymore. */
export function forgetRelease(runnerId: string, conversationId: string): void {
const left = pendingReleases(runnerId).filter((id) => id !== conversationId);
Expand Down
16 changes: 15 additions & 1 deletion packages/core/test/devices-e2e.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,8 @@ import { startDevices, stopDevices } from "../src/remote/devices";
import { startHumanTasks, stopHumanTasks } from "../src/services/humanTasks";
import { forwarding, groupRole, primaryRunnerId, resetGroupCache, runsShared } from "../src/remote/group";
import { startLinkServer, stopLinkServer } from "../src/remote/linkServer";
import { startRunners, stopRunners } from "../src/remote/runners";
import { pullFrom } from "../src/remote/replicaSync";
import { runnerInfoOf, startRunners, stopRunners } from "../src/remote/runners";
import { callTool } from "../src/mcp/tools";
import { setLogLevel } from "../src/log";
import { childEnv, sleep } from "../src/util";
Expand Down Expand Up @@ -274,6 +275,19 @@ describe("a device group, end to end", () => {
}
}, 60_000);

test("what the primary says about itself is asked again with each pull (a phone address it gets later reaches phones)", async () => {
rmSync(join(primaryDir, "fake-claude", "finish"), { force: true });
const started = await onPrimary<StartChatResult>("/api/chat", { method: "POST", body: JSON.stringify({ content: "WAIT_TO_FINISH while asked" }) });
const pid = primaryRunnerId()!;
await until(async () => {
await pullFrom(pid);
return (runnerInfoOf(pid)?.activeRuns ?? 0) > 0;
}, 20_000, "the primary's new answer about itself");
writeFileSync(join(primaryDir, "fake-claude", "finish"), "");
await until(async () => (await onPrimary<{ status: string }>(`/api/runs/${started.body.run.id}`)).body.status === "succeeded", 30_000, "the run to end");
rmSync(join(primaryDir, "fake-claude", "finish"), { force: true });
}, 60_000);

test("a question asked in a chat on this computer is answered through the primary", async () => {
const res = await here<StartChatResult>("/api/chat", { method: "POST", body: JSON.stringify({ content: "ASK_HUMAN please", runnerId: selfId }) });
const id = res.body.conversation.id;
Expand Down
Loading