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
1 change: 1 addition & 0 deletions apps/harness/config/runtime.exs
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ config :harness, HarnessWeb.Endpoint,

if config_env() == :prod do
trimmed_secret = String.trim(harness_secret || "")

if trimmed_secret == "" or trimmed_secret == "dev-harness-secret" do
raise "T3CODE_HARNESS_SECRET must be set to a non-default, non-empty value in production"
end
Expand Down
139 changes: 139 additions & 0 deletions apps/harness/lib/harness/dev/bridge_contract.ex
Original file line number Diff line number Diff line change
@@ -0,0 +1,139 @@
defmodule Harness.Dev.BridgeContract do
@moduledoc """
Structured bridge contract: Node ↔ Elixir channel messages.

This is the canonical source of truth for the Phoenix Channel surface
between Node.js (HarnessClientManager.ts) and Elixir (HarnessChannel).
"""

@doc """
Returns the full bridge contract as a structured map.
"""
def contract do
%{
title: "Node ↔ Elixir Bridge Contract",
description:
"Phoenix Channel over WebSocket. Node joins 'harness:lobby', " <>
"sends commands, receives push events. Socket URL: " <>
"ws://127.0.0.1:{port}/socket/websocket?secret={secret}&vsn=2.0.0",
node_to_elixir: node_to_elixir_events(),
elixir_to_node: elixir_to_node_events(),
lifecycle: lifecycle_notes()
}
end

defp node_to_elixir_events do
[
%{
event: "session.start",
params: ~w(threadId provider cwd model resumeCursor runtimeMode providerOptions),
required: ~w(threadId),
description: "Start a new provider session. Blocks until ready (up to 60s)."
},
%{
event: "session.sendTurn",
params: ~w(threadId input model effort interactionMode modelOptions),
required: ~w(threadId),
description: "Send a turn to an active session."
},
%{
event: "session.interrupt",
params: ~w(threadId turnId),
required: ~w(threadId),
description: "Interrupt the active turn."
},
%{
event: "session.respondToApproval",
params: ~w(threadId requestId decision),
required: ~w(threadId requestId decision),
description: "Respond to a tool approval request (approve/deny)."
},
%{
event: "session.respondToUserInput",
params: ~w(threadId requestId answers),
required: ~w(threadId requestId answers),
description: "Respond to a user input request."
},
%{
event: "session.stop",
params: ~w(threadId),
required: ~w(threadId),
description: "Stop a session and terminate its GenServer."
},
%{
event: "session.readThread",
params: ~w(threadId),
required: ~w(threadId),
description: "Read thread state from the provider."
},
%{
event: "session.rollbackThread",
params: ~w(threadId numTurns),
required: ~w(threadId numTurns),
description: "Rollback thread by N turns."
},
%{
event: "session.listSessions",
params: [],
required: [],
description: "List all active sessions."
},
%{
event: "session.stopAll",
params: [],
required: [],
description: "Stop all sessions."
},
%{
event: "provider.listModels",
params: ~w(provider),
required: ~w(provider),
description: "List models for a provider (cached in ETS, 10-min TTL)."
},
%{
event: "snapshot.get",
params: [],
required: [],
description: "Get current snapshot with all session states."
},
%{
event: "events.replay",
params: ~w(afterSeq),
required: ~w(afterSeq),
description: "Replay events since sequence number (WAL ring buffer, max 500 events)."
}
]
end

defp elixir_to_node_events do
[
%{
event: "harness.event",
fields: ~w(eventId threadId provider createdAt kind method payload seq),
kind_values: ~w(session notification request error),
description:
"Raw provider event with monotonic sequence number. " <>
"All provider lifecycle and content events flow through this push."
},
%{
event: "harness.session.changed",
fields: ~w(threadId session),
session_fields:
~w(threadId provider status model cwd runtimeMode activeTurn pendingRequests createdAt updatedAt),
description: "Pushed when session state changes (status, turn, requests)."
}
]
end

defp lifecycle_notes do
[
"Node connects to ws://127.0.0.1:{port}/socket/websocket with secret param",
"Node joins 'harness:lobby' topic",
"Heartbeat: Phoenix 'phoenix' topic, 30s interval",
"On reconnect: Node sends events.replay with lastSeenSeq to recover missed events",
"If afterSeq is too old (evicted from WAL), replay returns :gap — Node must full-resync via snapshot.get",
"Request timeout: 30s on Node side",
"Session start timeout: 60s (OpenCode server takes ~20s to boot)"
]
end
end
64 changes: 64 additions & 0 deletions apps/harness/lib/harness/dev/diagnostics_helpers.ex
Original file line number Diff line number Diff line change
@@ -0,0 +1,64 @@
defmodule Harness.Dev.DiagnosticsHelpers do
@moduledoc """
Shared helpers for provider session diagnostics.

Used by all provider GenServers in their `:get_diagnostics` handler
to produce a consistent, safe subset of internal state.
"""

@doc "Check if an Erlang Port is still alive."
def port_alive?(nil), do: false

def port_alive?(port) do
try do
Port.info(port) != nil
catch
_, _ -> false
end
end

@doc "Sanitize turn_state to only expose identifiers and timestamp."
def sanitize_turn_state(nil), do: nil

def sanitize_turn_state(turn) when is_map(turn),
do: Map.take(turn, [:id, :turn_id, :started_at])

def sanitize_turn_state(_), do: nil

@doc "Sanitize account to only expose type, plan_type, spark_enabled."
def sanitize_account(nil), do: nil

def sanitize_account(account) when is_map(account),
do: Map.take(account, [:type, :plan_type, :spark_enabled])

def sanitize_account(_), do: nil

@doc "Safely get process info for a pid. Returns nil if process is dead."
def process_info_safe(pid) when is_pid(pid) do
case Process.info(pid, [
:memory,
:heap_size,
:total_heap_size,
:message_queue_len,
:reductions
]) do
nil -> nil
info -> Map.new(info)
end
end

def process_info_safe(_), do: nil

@doc "Extract method names from a pending requests map (%{id => %{method: m, ...}})."
def pending_methods(pending) when is_map(pending) do
pending
|> Map.values()
|> Enum.map(fn
%{method: m} when is_binary(m) -> m
_ -> nil
end)
|> Enum.reject(&is_nil/1)
end

def pending_methods(_), do: []
end
149 changes: 149 additions & 0 deletions apps/harness/lib/harness/dev/doctor.ex
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
defmodule Harness.Dev.Doctor do
@moduledoc """
Transport-agnostic health probes for provider binaries and harness infrastructure.

Invariant: no harness state mutation. Side effects limited to diagnostic probes
(System.find_executable, System.cmd, Process.whereis).
"""

@check_timeout_ms 4_000

@provider_binaries %{
"codex" => %{binary: "codex", version_args: ["--version"]},
"claude" => %{binary: "claude", version_args: ["--version"]},
"cursor" => %{binary: "cursor", version_args: ["--version"]},
"opencode" => %{binary: "opencode", version_args: ["--version"]}
}

@doc """
Run all health checks. Returns overall status and per-check results.
"""
def full do
checks = %{
beam: check_beam(),
bridge: check_bridge(),
codex: check_binary("codex"),
claude: check_binary("claude"),
cursor: check_binary("cursor"),
opencode: check_binary("opencode")
}

overall =
if Enum.all?(Map.values(checks), &(&1.status in ["healthy", "not_installed"])) do
"healthy"
else
"degraded"
end

%{overall: overall, checks: checks, timestamp: now_ms()}
end

@doc """
Run a single health check by target name.
"""
def check(target) when target in ~w(codex claude cursor opencode) do
{:ok, check_binary(target)}
end

def check("bridge"), do: {:ok, check_bridge()}
def check("beam"), do: {:ok, check_beam()}

def check(other),
do: {:error, "Unknown target: #{other}. Valid: codex, claude, cursor, opencode, bridge, beam"}

# --- Binary checks ---

defp check_binary(provider) do
case Map.get(@provider_binaries, provider) do
nil ->
%{status: "error", detail: "Unknown provider: #{provider}"}

%{binary: name, version_args: args} ->
do_check_binary(name, args)
end
end

defp do_check_binary(name, args) do
case System.find_executable(name) do
nil ->
%{status: "not_installed", binary: nil, version: nil, detail: "#{name} not found in PATH"}

path ->
case run_version_check(path, args) do
{:ok, version} ->
%{status: "healthy", binary: path, version: version, detail: nil}

{:error, reason} ->
%{status: "degraded", binary: path, version: nil, detail: reason}
end
end
end

defp run_version_check(binary, args) do
task =
Task.async(fn ->
System.cmd(binary, args, stderr_to_stdout: true)
end)

case Task.yield(task, @check_timeout_ms) || Task.shutdown(task) do
{:ok, {output, 0}} ->
version = parse_version(output)
{:ok, version || String.trim(output)}

{:ok, {output, code}} ->
{:error, "Exit code #{code}: #{String.slice(String.trim(output), 0, 200)}"}

{:exit, reason} ->
{:error, "Task exited: #{inspect(reason)}"}

nil ->
{:error, "Version check timed out after #{@check_timeout_ms}ms"}
end
rescue
Comment on lines +88 to +102

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

🧩 Analysis chain

🌐 Web query:

In Elixir, what are the default timeout and return semantics of Task.shutdown/2, and can Task.yield(task, 4000) || Task.shutdown(task) wait longer than 4000ms?

💡 Result:

In Elixir, Task.shutdown/2 has a default timeout of 5000 milliseconds (5 seconds). It unlinks from the task process, sends a :shutdown exit signal, waits up to the timeout for the task to terminate (checking via monitor), and then brutally kills it if needed. Its return semantics match Task.yield/2: {:ok, term} if a reply is received, {:exit, reason} if the task exits, or nil if the timeout is reached without a reply or DOWN message. The expression Task.yield(task, 4000) || Task.shutdown(task) can wait longer than 4000ms. Task.yield/2 waits up to 4000ms. If it returns nil (no reply in time), Task.shutdown/2 (with default 5000ms timeout) is called next, which can block for up to another 5000ms waiting for shutdown completion, for a total potential wait of up to 9000ms. This pattern ensures late replies (after yield timeout but before shutdown starts) are captured, as documented.

Citations:


Missing {:exit, reason} pattern and timeout can exceed 4000ms.

The expression Task.yield(task, @check_timeout_ms) || Task.shutdown(task) on line 88 doesn't explicitly handle {:exit, reason} outcomes from either Task.yield/2 or Task.shutdown/2, which causes a match error if a task exits. Additionally, the total timeout can reach 9000ms (4000ms from yield + 5000ms default shutdown timeout), not just 4000ms as the error message indicates.

💡 Suggested fix
-    case Task.yield(task, `@check_timeout_ms`) || Task.shutdown(task) do
+    case Task.yield(task, `@check_timeout_ms`) do
       {:ok, {output, 0}} ->
         version = parse_version(output)
         {:ok, version || String.trim(output)}
 
       {:ok, {output, code}} ->
         {:error, "Exit code #{code}: #{String.slice(String.trim(output), 0, 200)}"}
 
+      {:exit, reason} ->
+        {:error, "Version check failed: #{inspect(reason)}"}
+
       nil ->
+        _ = Task.shutdown(task, :brutal_kill)
         {:error, "Version check timed out after #{`@check_timeout_ms`}ms"}
     end
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
case Task.yield(task, @check_timeout_ms) || Task.shutdown(task) do
{:ok, {output, 0}} ->
version = parse_version(output)
{:ok, version || String.trim(output)}
{:ok, {output, code}} ->
{:error, "Exit code #{code}: #{String.slice(String.trim(output), 0, 200)}"}
nil ->
{:error, "Version check timed out after #{@check_timeout_ms}ms"}
end
rescue
case Task.yield(task, `@check_timeout_ms`) do
{:ok, {output, 0}} ->
version = parse_version(output)
{:ok, version || String.trim(output)}
{:ok, {output, code}} ->
{:error, "Exit code #{code}: #{String.slice(String.trim(output), 0, 200)}"}
{:exit, reason} ->
{:error, "Version check failed: #{inspect(reason)}"}
nil ->
_ = Task.shutdown(task, :brutal_kill)
{:error, "Version check timed out after #{`@check_timeout_ms`}ms"}
end
rescue
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@apps/harness/lib/harness/dev/doctor.ex` around lines 88 - 99, The case
handling around Task.yield(task, `@check_timeout_ms`) || Task.shutdown(task) must
explicitly handle {:exit, reason} and accurately reflect the real timeout;
update the case in the function (the block using Task.yield/2, Task.shutdown/2
and parse_version) to match an {:exit, reason} tuple and return an {:error,
"Exited: #{inspect(reason)}"} (or similar) instead of letting a match error
happen, and either call Task.shutdown(task, 0) or pass an explicit shutdown
timeout and update the timeout message to report the combined yield+shutdown
duration (use `@check_timeout_ms` plus the shutdown timeout) so the error text is
correct. Ensure parse_version handling remains for {:ok, {output, 0}} and other
{:ok, {output, code}} branches.

e -> {:error, inspect(e)}
end

defp parse_version(output) do
case Regex.run(~r/\bv?(\d+\.\d+(?:\.\d+)?(?:-[0-9A-Za-z.-]+)?)\b/, output) do
[_, version] -> version
_ -> nil
end
end

# --- Infrastructure checks ---

defp check_bridge do
endpoint_running = Process.whereis(HarnessWeb.Endpoint) != nil
pubsub_alive = Process.whereis(Harness.PubSub) != nil
registry_alive = Process.whereis(Harness.SessionRegistry) != nil
supervisor_alive = Process.whereis(Harness.SessionSupervisor) != nil
snapshot_alive = Process.whereis(Harness.SnapshotServer) != nil

all_alive =
endpoint_running and pubsub_alive and registry_alive and supervisor_alive and snapshot_alive

%{
status: if(all_alive, do: "healthy", else: "degraded"),
endpoint_running: endpoint_running,
pubsub_alive: pubsub_alive,
registry_alive: registry_alive,
supervisor_alive: supervisor_alive,
snapshot_server_alive: snapshot_alive
}
end

defp check_beam do
memory = :erlang.memory()
total_mb = Float.round(memory[:total] / (1024 * 1024), 1)

%{
status: "healthy",
process_count: :erlang.system_info(:process_count),
total_memory_mb: total_mb,
scheduler_count: :erlang.system_info(:schedulers_online),
otp_release: to_string(:erlang.system_info(:otp_release))
}
end

defp now_ms, do: System.system_time(:millisecond)
end
Loading
Loading