fix(elixir): ignore codex stream noise in dashboard events

Summary:
- only emit malformed Codex events for JSON-like protocol frames that
  fail to decode
- keep logging non-JSON stream output without surfacing it to the
  orchestrator dashboard
- add regression coverage for stderr noise and malformed protocol
  frames

Rationale:
- Codex can write diagnostics to stderr during healthy turns, and the
  merged stdio stream was incorrectly surfacing those lines as
  malformed JSON events in the Symphony UI
- keeping malformed detection for JSON-looking frames preserves signal
  for actual protocol corruption while removing false positives

Tests:
- mise exec -- mix format lib/symphony_elixir/codex/app_server.ex
  test/symphony_elixir/app_server_test.exs
- MIX_ENV=test mise exec -- mix run --no-start -e
  'Application.put_env(:symphony_elixir, :workflow_file_path,
  System.fetch_env!("SYMPHONY_TEST_WORKFLOW")); Mix.Task.run("test",
  ["test/symphony_elixir/app_server_test.exs"])'

Co-authored-by: Codex <noreply@openai.com>

Author: kevinw-openai

elixir/lib/symphony_elixir/codex/app_server.ex

@@ -381,15 +381,17 @@ defmodule SymphonyElixir.Codex.AppServer do
       {:error, _reason} ->
         log_non_json_stream_line(payload_string, "turn stream")
 
-        emit_message(
-          on_message,
-          :malformed,
-          %{
-            payload: payload_string,
-            raw: payload_string
-          },
-          metadata_from_message(port, %{raw: payload_string})
-        )
+        if protocol_message_candidate?(payload_string) do
+          emit_message(
+            on_message,
+            :malformed,
+            %{
+              payload: payload_string,
+              raw: payload_string
+            },
+            metadata_from_message(port, %{raw: payload_string})
+          )
+        end
 
         receive_loop(port, on_message, timeout_ms, "", tool_executor, auto_approve_requests)
     end
@@ -895,6 +897,13 @@ defmodule SymphonyElixir.Codex.AppServer do
     end
   end
 
+  defp protocol_message_candidate?(data) do
+    data
+    |> to_string()
+    |> String.trim_leading()
+    |> String.starts_with?("{")
+  end
+
   defp issue_context(%{id: issue_id, identifier: identifier}) do
     "issue_id=#{issue_id} issue_identifier=#{identifier}"
   end

elixir/test/symphony_elixir/app_server_test.exs

@@ -1189,14 +1189,91 @@ defmodule SymphonyElixir.AppServerTest do
         labels: ["backend"]
       }
 
+      test_pid = self()
+      on_message = fn message -> send(test_pid, {:app_server_message, message}) end
+
       log =
         capture_log(fn ->
-          assert {:ok, _result} = AppServer.run(workspace, "Capture stderr log", issue)
+          assert {:ok, _result} =
+                   AppServer.run(workspace, "Capture stderr log", issue, on_message: on_message)
         end)
 
+      assert_received {:app_server_message, %{event: :turn_completed}}
+      refute_received {:app_server_message, %{event: :malformed}}
       assert log =~ "Codex turn stream output: warning: this is stderr noise"
     after
       File.rm_rf(test_root)
     end
   end
+
+  test "app server emits malformed events for JSON-like protocol lines that fail to decode" do
+    test_root =
+      Path.join(
+        System.tmp_dir!(),
+        "symphony-elixir-app-server-malformed-protocol-#{System.unique_integer([:positive])}"
+      )
+
+    try do
+      workspace_root = Path.join(test_root, "workspaces")
+      workspace = Path.join(workspace_root, "MT-93")
+      codex_binary = Path.join(test_root, "fake-codex")
+      File.mkdir_p!(workspace)
+
+      File.write!(codex_binary, """
+      #!/bin/sh
+      count=0
+      while IFS= read -r line; do
+        count=$((count + 1))
+
+        case "$count" in
+          1)
+            printf '%s\\n' '{"id":1,"result":{}}'
+            ;;
+          2)
+            printf '%s\\n' '{"id":2,"result":{"thread":{"id":"thread-93"}}}'
+            ;;
+          3)
+            printf '%s\\n' '{"id":3,"result":{"turn":{"id":"turn-93"}}}'
+            ;;
+          4)
+            printf '%s\\n' '{"method":"turn/completed"'
+            printf '%s\\n' '{"method":"turn/completed"}'
+            exit 0
+            ;;
+          *)
+            exit 0
+            ;;
+        esac
+      done
+      """)
+
+      File.chmod!(codex_binary, 0o755)
+
+      write_workflow_file!(Workflow.workflow_file_path(),
+        workspace_root: workspace_root,
+        codex_command: "#{codex_binary} app-server"
+      )
+
+      issue = %Issue{
+        id: "issue-malformed-protocol",
+        identifier: "MT-93",
+        title: "Malformed protocol frame",
+        description: "Ensure malformed JSON-like frames are surfaced to the orchestrator",
+        state: "In Progress",
+        url: "https://example.org/issues/MT-93",
+        labels: ["backend"]
+      }
+
+      test_pid = self()
+      on_message = fn message -> send(test_pid, {:app_server_message, message}) end
+
+      assert {:ok, _result} =
+               AppServer.run(workspace, "Capture malformed protocol line", issue, on_message: on_message)
+
+      assert_received {:app_server_message, %{event: :malformed, payload: "{\"method\":\"turn/completed\""}}
+      assert_received {:app_server_message, %{event: :turn_completed}}
+    after
+      File.rm_rf(test_root)
+    end
+  end
 end