shifathosn
/
rea
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(hopper): report bridge progress diagnostics

This commit is contained in:
morluto 2026-07-25 14:29:02 +00:00
parent 305e72995b
commit 7a43420029
12 changed files with 473 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
bridge/hopper_bridge.py
View File

@ -1046,29 +1046,65 @@ def _serve_connection(connection):
if not line or len(line) > MAX_LINE_BYTES:
break
request_id = None
authenticated = False
should_stop = False
try:
request = json.loads(line.decode("utf-8"))
if set(request) != {"id", "token", "method", "params"}:
raise ValueError("Invalid bridge request shape")
request_id = request["id"]
if isinstance(request_id, bool) or not isinstance(request_id, int) or request_id < 0:
raise ValueError("Invalid bridge request id")
if not isinstance(request["token"], str) or not hmac.compare_digest(request["token"], REA_TOKEN):
raise PermissionError("Invalid bridge capability")
authenticated = True
_write_message(file, {"id": request_id, "event": {
"type": "progress",
"phase": "hopper_bridge",
"completed": 0,
"total": 1,
"message": "Hopper bridge started request",
}})
result = _dispatch(request["method"], request["params"])
should_stop = request["method"] == "shutdown_document" or (
request["method"] == "shutdown" and not result.get("cleanup_required", False)
)
response = {"id": request_id, "result": _json_safe(result)}
safe_result = _json_safe(result)
_write_message(file, {"id": request_id, "event": {
"type": "progress",
"phase": "hopper_bridge",
"completed": 1,
"total": 1,
"message": "Hopper bridge completed request",
"terminal": True,
}})
response = {"id": request_id, "result": safe_result}
except Exception as error:
response = {"id": request_id if isinstance(request_id, int) else 0, "error": {"code": -32000, "message": str(error)[:512], "type": _diagnostic_type(error)}}
file.write((json.dumps(response, separators=(",", ":")) + "\n").encode("utf-8"))
file.flush()
diagnostic = _safe_diagnostic(error)
if authenticated:
_write_message(file, {"id": request_id, "event": {
"type": "diagnostic",
"error": diagnostic,
}})
response_id = request_id if (
isinstance(request_id, int)
and not isinstance(request_id, bool)
and request_id >= 0
) else 0
response = {"id": response_id, "error": diagnostic}
_write_message(file, response)
if should_stop:
break
file.close()
connection.close()
def _write_message(file, message):
"""Write and flush one compact bridge message."""
file.write((json.dumps(message, separators=(",", ":")) + "\n").encode("utf-8"))
file.flush()
def _diagnostic_type(error):
if isinstance(error, CapabilityUnavailableError):
return "capability_unavailable"
@ -1079,6 +1115,20 @@ def _diagnostic_type(error):
return "bridge_exception"
def _safe_diagnostic(error):
"""Project one exception without retaining provider or credential text."""
diagnostic_type = _diagnostic_type(error)
if diagnostic_type == "capability_unavailable":
message = str(error)[:512]
elif diagnostic_type == "authorization":
message = "Invalid bridge capability"
elif diagnostic_type == "invalid_request":
message = "Invalid Hopper bridge request"
else:
message = "%s: Hopper bridge operation failed" % type(error).__name__[:128]
return {"code": -32000, "message": message, "type": diagnostic_type}
def _run():
"""Own a permission-restricted, single-client Unix socket for this bridge."""
if os.path.exists(REA_SOCKET):

20
docs/installation.md
View File

@ -145,6 +145,26 @@ host mount and the rest of `/tmp` remain unchanged; this fallback never invokes
`sudo`. `rea doctor --provider hopper --detail full --json` reports the selected
strategy and both host and effective mount facts.
### Hopper in CI
REA's unattended Linux path is validated against Hopper's offered demo mode.
A paid license is not required for that path, and REA does not read, install, or
automate license credentials. Licensed Hopper installations remain supported,
but license activation is an operator-owned prerequisite rather than part of
REA setup.
macOS requires Hopper's first-run UI to be completed in the same user session
that will run REA: choose the demo mode or activate an existing license before
starting an unattended job. Ephemeral macOS runners therefore need a
pre-provisioned user session or a deliberate interactive bootstrap step.
When Hopper cannot start in CI, run
`rea doctor --provider hopper --detail full --json` in the failing runner.
Structured failures distinguish private-display dependencies, an unsupported
demo dialog or build, process-ownership conflicts, and an early lifecycle exit.
Apply the reported remediation rather than exposing the runner's desktop,
copying license secrets into logs, or killing unrelated Hopper processes.
## Ghidra
REA's Ghidra provider is bring-your-own and supports Linux x64 with the exact

2
src/domain/hopperStartupFailure.ts
View File

@ -65,7 +65,7 @@ const failures = [
[
75,
"hopper_exited_during_startup",
"Hopper exited before the REA bridge was ready. Run rea doctor and retry the analysis.",
"Hopper exited before the REA bridge was ready. In CI, run rea doctor --provider hopper --detail full --json; if Hopper showed a UI or license prompt, complete it in the runner user session, otherwise close stale Hopper sessions and retry.",
],
[
76,

35
src/hopper/HopperClient.ts
View File

@ -42,7 +42,11 @@ import {
type HopperRequestActivity,
} from "./HopperRequestQueue.js";
import { HopperResponseStream } from "./HopperResponseStream.js";
import { responseResult } from "./protocol.js";
import {
type HopperBridgeEvent,
type HopperBridgeMessage,
responseResult,
} from "./protocol.js";
export type { HopperServerInfo } from "./HopperSessionValues.js";
@ -125,8 +129,7 @@ export class HopperClient {
},
);
this.#responses = new HopperResponseStream({
accept: (response) =>
this.#requests.accept(response.id, responseResult(response)),
accept: (message) => this.#acceptBridgeMessage(message),
hasQueued: (id) => this.#requests.hasQueued(id),
nextRequestId: () => this.#nextId,
abort: (message, cause) => this.#abortProtocol(message, cause),
@ -450,6 +453,32 @@ export class HopperClient {
socket.on("close", this.#onSocketClose);
}
#acceptBridgeMessage(message: HopperBridgeMessage): boolean {
if (!("event" in message))
return this.#requests.accept(message.id, responseResult(message));
if (!this.#requests.acceptEvent(message)) return false;
this.#emitBridgeDiagnostic(message);
return true;
}
#emitBridgeDiagnostic(message: HopperBridgeEvent): void {
if (message.event.type !== "diagnostic") return;
try {
this.#options.onDiagnostic?.({
type: "bridge-diagnostic",
request_id: message.id,
code: message.event.error.code,
category: message.event.error.type,
message: message.event.error.message,
});
} catch {
this.#logger.warn(
{ requestId: message.id },
"Hopper bridge diagnostic consumer rejected an event",
);
}
}
#detachSocket(socket: Socket): void {
socket.off("data", this.#onSocketData);
socket.off("error", this.#onSocketError);

8
src/hopper/HopperDiagnostics.ts
View File

@ -1,3 +1,4 @@
import type { HopperDiagnosticType } from "../domain/errors.js";
import type { ProcessCleanupResult } from "../process/ProcessOwnership.js";
import type { ProviderProcessStopResult } from "../process/ProviderProcess.js";
import type { BridgeLaunch } from "./BridgeLauncher.js";
@ -6,6 +7,13 @@ import type { BridgeLaunch } from "./BridgeLauncher.js";
export type HopperDiagnostic =
| { readonly type: "launcher-stderr"; readonly bytes: number }
| { readonly type: "launcher-exit"; readonly code: number | null }
| {
readonly type: "bridge-diagnostic";
readonly request_id: number;
readonly code: number;
readonly category: HopperDiagnosticType;
readonly message: string;
}
| {
readonly type: "owned-shutdown";
readonly status: ProviderProcessStopResult["status"];

52
src/hopper/HopperRequestQueue.ts
View File

@ -1,6 +1,10 @@
import { performance } from "node:perf_hooks";
import type { ProgressReporter } from "../application/ProgressReporter.js";
import type {
ProgressReporter,
ProgressUpdate,
} from "../application/ProgressReporter.js";
import type { HopperBridgeEvent } from "./protocol.js";
import {
HopperCancelledError,
type HopperError,
@ -127,6 +131,28 @@ export class HopperRequestQueue {
return true;
}
/** Forward one correlated bridge event without releasing the wire request. */
acceptEvent(event: HopperBridgeEvent): boolean {
const entry = this.#active;
if (entry === undefined || entry.id !== event.id) return false;
if (event.event.type === "progress" && !entry.callerSettled) {
if (event.event.terminal === true && entry.heartbeat !== undefined) {
clearInterval(entry.heartbeat);
entry.heartbeat = undefined;
}
this.#reportUpdate(entry, {
phase: event.event.phase,
completed: event.event.completed,
total: event.event.total,
message: event.event.message,
...(event.event.terminal === undefined
? {}
: { terminal: event.event.terminal }),
});
}
return true;
}
/** Fail the active wire request and every request still waiting in the FIFO. */
failAll(error: HopperError): void {
const active = this.#active;
@ -267,13 +293,21 @@ export class HopperRequestQueue {
}
#report(entry: QueuedRequest, message: string): void {
void entry.progress
?.report({
phase: "hopper_request",
completed: 0,
total: 1,
message,
})
.catch(() => undefined);
this.#reportUpdate(entry, {
phase: "hopper_request",
completed: 0,
total: 1,
message,
});
}
#reportUpdate(entry: QueuedRequest, update: ProgressUpdate): void {
try {
void Promise.resolve(entry.progress?.report(update)).catch(
() => undefined,
);
} catch {
// Progress observation cannot change the request outcome.
}
}
}

9
src/hopper/HopperResponseStream.ts
View File

@ -1,9 +1,12 @@
import { parseResponseLine, type HopperResponse } from "./protocol.js";
import {
parseBridgeMessageLine,
type HopperBridgeMessage,
} from "./protocol.js";
const MAX_LINE_BYTES = 10 * 1024 * 1024;
export interface HopperResponseStreamOptions {
readonly accept: (response: HopperResponse) => boolean;
readonly accept: (message: HopperBridgeMessage) => boolean;
readonly hasQueued: (id: number) => boolean;
readonly nextRequestId: () => number;
readonly abort: (message: string, cause?: Error) => void;
@ -38,7 +41,7 @@ export class HopperResponseStream {
}
#acceptLine(line: string): boolean {
const parsed = parseResponseLine(line);
const parsed = parseBridgeMessageLine(line);
if (!parsed.ok) {
this.options.abort(parsed.error.message, parsed.error);
return false;

101
src/hopper/protocol.ts
View File

@ -4,35 +4,84 @@ import { HopperProtocolError, HopperRemoteError } from "../domain/errors.js";
import { jsonValueSchema, type JsonValue } from "../domain/jsonValue.js";
import { err, ok, type Result } from "../domain/result.js";
const remoteErrorSchema = z
.object({
code: z.number().int(),
message: z.string().max(512),
type: z
.enum([
"remote",
"authorization",
"invalid_request",
"capability_unavailable",
"bridge_exception",
])
.default("remote"),
})
.strict();
const responseSchema = z.union([
z.object({
id: z.number().int().nonnegative(),
result: jsonValueSchema,
}),
z.object({
id: z.number().int().nonnegative(),
error: z.object({
code: z.number().int(),
message: z.string(),
type: z
.enum([
"remote",
"authorization",
"invalid_request",
"capability_unavailable",
"bridge_exception",
])
.default("remote"),
}),
}),
z
.object({
id: z.number().int().nonnegative(),
result: jsonValueSchema,
})
.strict(),
z
.object({
id: z.number().int().nonnegative(),
error: remoteErrorSchema,
})
.strict(),
]);
export type HopperResponse = z.infer<typeof responseSchema>;
const eventSchema = z
.object({
id: z.number().int().nonnegative(),
event: z.discriminatedUnion("type", [
z
.object({
type: z.literal("progress"),
phase: z.string().min(1).max(64),
completed: z.number().nonnegative().finite(),
total: z.number().nonnegative().finite().nullable(),
message: z.string().min(1).max(512),
terminal: z.boolean().optional(),
})
.strict(),
z
.object({
type: z.literal("diagnostic"),
error: remoteErrorSchema,
})
.strict(),
]),
})
.strict()
.superRefine((message, context) => {
if (
message.event.type === "progress" &&
message.event.total !== null &&
message.event.completed > message.event.total
) {
context.addIssue({
code: "custom",
message: "Progress total cannot be less than completed",
path: ["event", "total"],
});
}
});
/** Parse one complete Hopper NDJSON response line. */
export const parseResponseLine = (
const messageSchema = z.union([responseSchema, eventSchema]);
export type HopperResponse = z.infer<typeof responseSchema>;
export type HopperBridgeEvent = z.infer<typeof eventSchema>;
export type HopperBridgeMessage = z.infer<typeof messageSchema>;
/** Parse one complete Hopper NDJSON response or event line. */
export const parseBridgeMessageLine = (
line: string,
): Result<HopperResponse, HopperProtocolError> => {
): Result<HopperBridgeMessage, HopperProtocolError> => {
let decoded: unknown;
try {
decoded = JSON.parse(line);
@ -42,12 +91,12 @@ export const parseResponseLine = (
);
}
const parsed = responseSchema.safeParse(decoded);
const parsed = messageSchema.safeParse(decoded);
return parsed.success
? ok(parsed.data)
: err(
new HopperProtocolError(
"Hopper returned a response outside the bridge contract",
"Hopper returned a message outside the bridge contract",
{ cause: parsed.error },
),
);

82
tests/fixtures/fakeHopper.mjs vendored
View File

@ -16,6 +16,43 @@ const capabilityUnavailableResult = (id) => ({
type: "capability_unavailable",
},
});
const bridgeEventFixtureMessages = (request) => {
const progress = {
type: "progress",
phase: "hopper_bridge",
completed: 0,
total: 1,
message: "Hopper bridge started request",
};
switch (request.method) {
case "wrong_event_id":
return [{ id: request.id + 100, event: progress }];
case "malformed_event":
return [
{
id: request.id,
event: {
...progress,
completed: 2,
message: "invalid progress",
},
},
];
case "remote_error": {
const error = {
code: -32001,
message: "safe fake failure",
type: "bridge_exception",
};
return [
{ id: request.id, event: { type: "diagnostic", error } },
{ id: request.id, error },
];
}
default:
return undefined;
}
};
const enhancedFixtureResult = (method) => {
switch (method) {
case "list_segments":
@ -109,6 +146,33 @@ const enhancedFixtureResult = (method) => {
}
};
const sendEchoWithProgress = (send, request) => {
send({
id: request.id,
event: {
type: "progress",
phase: "hopper_bridge",
completed: 0,
total: 1,
message: "Hopper bridge started request",
},
});
setTimeout(() => {
send({
id: request.id,
event: {
type: "progress",
phase: "hopper_bridge",
completed: 1,
total: 1,
message: "Hopper bridge completed request",
terminal: true,
},
});
send({ id: request.id, result: request.params ?? {} });
}, request.params?.delay ?? 0);
};
const server = createServer((socket) => {
socket.on("error", () => undefined);
socket.setEncoding("utf8");
@ -125,6 +189,7 @@ const server = createServer((socket) => {
let newline = buffer.indexOf("\n");
while (newline >= 0) {
const request = JSON.parse(buffer.slice(0, newline));
const bridgeEventMessages = bridgeEventFixtureMessages(request);
buffer = buffer.slice(newline + 1);
if (request.token !== token) {
send({ id: request.id, error: { code: -32001, message: "bad token" } });
@ -152,15 +217,8 @@ const server = createServer((socket) => {
socket.write("{not-json}\n");
} else if (request.method === "wrong_id") {
send({ id: request.id + 100, result: {} });
} else if (request.method === "remote_error") {
send({
id: request.id,
error: {
code: -32001,
message: "safe fake failure",
type: "bridge_exception",
},
});
} else if (bridgeEventMessages !== undefined) {
for (const message of bridgeEventMessages) send(message);
} else if (request.method === "capability_unavailable") {
send(capabilityUnavailableResult(request.id));
} else if (request.method === "current_document") {
@ -194,11 +252,7 @@ const server = createServer((socket) => {
},
});
} else {
const delay = request.params?.delay ?? 0;
setTimeout(
() => send({ id: request.id, result: request.params ?? {} }),
delay,
);
sendEchoWithProgress(send, request);
}
newline = buffer.indexOf("\n");
}

32
tests/fixtures/hopperBridgeFacadeProbe.py vendored
View File

@ -2,7 +2,9 @@
import json
from pathlib import Path
import socket
import sys
import threading
class FakeDocument:
@ -74,6 +76,34 @@ def main():
"message": str(error),
}
bridge["REA_TOKEN"] = "probe-token"
bridge["_dispatch"] = lambda method, params: (
(_ for _ in ()).throw(RuntimeError("credential=supersecret"))
if method == "fail"
else params
)
server_socket, client_socket = socket.socketpair()
worker = threading.Thread(
target=bridge["_serve_connection"], args=(server_socket,)
)
worker.start()
client_file = client_socket.makefile("rwb")
client_file.write(
b'{"id":7,"token":"probe-token","method":"fail","params":{}}\n'
)
client_file.flush()
bridge_messages = [
json.loads(client_file.readline().decode("utf-8")) for _ in range(3)
]
client_file.write(
b'{"id":-1,"token":"probe-token","method":"echo","params":{}}\n'
)
client_file.flush()
invalid_id_response = json.loads(client_file.readline().decode("utf-8"))
client_file.close()
client_socket.close()
worker.join(timeout=1)
print(
json.dumps(
{
@ -82,6 +112,8 @@ def main():
"current_address": current_address,
"session_document": selected,
"analysis_guard": analysis_guard,
"bridge_messages": bridge_messages,
"invalid_id_response": invalid_id_response,
},
sort_keys=True,
)

40
tests/hopperBridgeFacade.test.ts
View File

@ -24,6 +24,45 @@ const probeResultSchema = z.strictObject({
diagnostic_type: z.literal("capability_unavailable"),
message: z.string(),
}),
bridge_messages: z.tuple([
z.strictObject({
id: z.literal(7),
event: z.strictObject({
type: z.literal("progress"),
phase: z.literal("hopper_bridge"),
completed: z.literal(0),
total: z.literal(1),
message: z.literal("Hopper bridge started request"),
}),
}),
z.strictObject({
id: z.literal(7),
event: z.strictObject({
type: z.literal("diagnostic"),
error: z.strictObject({
code: z.literal(-32000),
message: z.literal("RuntimeError: Hopper bridge operation failed"),
type: z.literal("bridge_exception"),
}),
}),
}),
z.strictObject({
id: z.literal(7),
error: z.strictObject({
code: z.literal(-32000),
message: z.literal("RuntimeError: Hopper bridge operation failed"),
type: z.literal("bridge_exception"),
}),
}),
]),
invalid_id_response: z.strictObject({
id: z.literal(0),
error: z.strictObject({
code: z.literal(-32000),
message: z.literal("Invalid Hopper bridge request"),
type: z.literal("invalid_request"),
}),
}),
});
describe("Hopper API facade", () => {
@ -38,6 +77,7 @@ describe("Hopper API facade", () => {
},
);
const result = probeResultSchema.parse(JSON.parse(stdout));
expect(stdout).not.toContain("supersecret");
expect(result.analysis_guard.message).toContain(
"requires completed Hopper background analysis",
);

95
tests/hopperClient.test.ts
View File

@ -14,6 +14,7 @@ import type {
BridgeSession,
} from "../src/hopper/BridgeLauncher.js";
import { HopperClient } from "../src/hopper/HopperClient.js";
import type { HopperDiagnostic } from "../src/hopper/HopperDiagnostics.js";
import { LINUX_PRIVATE_DISPLAY_DIAGNOSTIC_PREFIX } from "../src/hopper/LinuxPrivateDisplayDiagnostic.js";
import { providerCleanupFailure } from "../src/hopper/HopperDiagnostics.js";
import { cleanupOwnedProcessGroup } from "../src/process/ProcessOwnership.js";
@ -296,6 +297,8 @@ describe("HopperClient", () => {
it.each([
["malformed", "HopperProtocolError"],
["wrong_id", "HopperProtocolError"],
["wrong_event_id", "HopperProtocolError"],
["malformed_event", "HopperProtocolError"],
["remote_error", "HopperRemoteError"],
["hang", "HopperTimeoutError"],
["exit", "HopperProcessError"],
@ -307,7 +310,15 @@ describe("HopperClient", () => {
});
it("preserves a sanitized bridge exception diagnostic", async () => {
const client = await startClient();
const diagnostics: HopperDiagnostic[] = [];
const client = new HopperClient({
launcher: new FixtureLauncher(),
requestTimeoutMs: 100,
startupTimeoutMs: 1_000,
onDiagnostic: (event) => diagnostics.push(event),
});
clients.push(client);
await expect(client.start()).resolves.toMatchObject({ ok: true });
const result = await client.callTool("remote_error");
expect(result.ok).toBe(false);
if (!result.ok)
@ -315,6 +326,66 @@ describe("HopperClient", () => {
_tag: "HopperRemoteError",
diagnosticType: "bridge_exception",
});
expect(diagnostics).toContainEqual({
type: "bridge-diagnostic",
request_id: 2,
code: -32001,
category: "bridge_exception",
message: "safe fake failure",
});
});
it("forwards correlated Python progress before the terminal response", async () => {
const client = await startClient();
const updates: Array<{
readonly phase: string;
readonly completed: number;
readonly terminal?: boolean;
}> = [];
const result = await client.callTool(
"echo",
{ value: "progress" },
{
progress: {
report: (update) => {
updates.push(update);
return Promise.resolve();
},
},
},
);
expect(result).toEqual({ ok: true, value: { value: "progress" } });
expect(updates).toEqual(
expect.arrayContaining([
expect.objectContaining({
phase: "hopper_bridge",
completed: 0,
}),
expect.objectContaining({
phase: "hopper_bridge",
completed: 1,
terminal: true,
}),
]),
);
});
it("isolates synchronous progress observer failures from bridge responses", async () => {
const client = await startClient();
await expect(
client.callTool(
"echo",
{ value: "alive" },
{
progress: {
report: () => {
throw new Error("observer failed");
},
},
},
),
).resolves.toEqual({ ok: true, value: { value: "alive" } });
});
it("projects a missing bridge API as typed capability unavailability", async () => {
@ -460,6 +531,28 @@ describe("HopperClient", () => {
},
);
it("projects actionable CI remediation for Hopper UI, license, and lifecycle failures", async () => {
const client = new HopperClient({
launcher: new ExitingLauncher(75),
startupTimeoutMs: 10_000,
});
clients.push(client);
const result = await client.start();
expect(result.ok).toBe(false);
if (result.ok) return;
const projected = projectAnalysisError(result.error);
expect(projected.message).toContain(
"rea doctor --provider hopper --detail full --json",
);
expect(projected.message).toContain("UI or license prompt");
expect(projected.message).toContain("close stale Hopper sessions");
expect(projected.details).toMatchObject({
failure_code: "hopper_exited_during_startup",
exit_code: 75,
});
});
it("preserves the bounded private-display diagnostic from an adapter exit", async () => {
const client = new HopperClient({
launcher: new DiagnosticExitingLauncher(),