diagnose(shell): trace input acceptance and writes (#73)

experiments/shell/queues.hpp

@@ -63,11 +63,15 @@ class InputQueue {
         size_t queued_bytes = 0, in_flight_bytes = 0, high_water_bytes = 0, rejected = 0;
         bool stopping = false, started = false, joined = false, failed = false;
     };
+    // Opt-in trace of input acceptance and of the bytes that reached the
+    // session, used to tell a stalled input path from a stalled output path.
+    using TraceHook = void (*)(const void* owner, bool written, size_t bytes, bool accepted);
     static constexpr size_t capacity = 16 * 1024;
     // The legacy high water is max(queued keystroke bytes, individual paste
     // chunk bytes); it is not queued + in-flight or a total retained budget.
     // Legacy counters are safe to read only after the writer has joined.
     size_t high_water = 0, rejected = 0;
+    TraceHook trace_hook_ = nullptr;
     explicit InputQueue(tf_conpty::Session& session) : session_(session) {}
     ~InputQueue() { stop(); }
     Snapshot snapshot() const {
@@ -101,6 +105,7 @@ class InputQueue {
                         session_.send(bytes, [&](size_t written) { writing_paste->acknowledge_written(written); });
                         writing_paste->finish_chunk();
                     } else session_.send(bytes);
+                    if (trace_hook_) trace_hook_(this, true, bytes.size(), true);
                     bytes.clear(); // Payload is released; string allocation is not measured here.
                     lock.lock(); in_flight_bytes_ = 0;
                 }
@@ -120,13 +125,21 @@ class InputQueue {
         });
         started_ = true;
     }
+    void set_trace_hook(TraceHook hook) { trace_hook_ = hook; }
     bool send(const std::string& bytes) {
-        std::lock_guard<std::mutex> lock(mutex_);
-        if (error_) std::rethrow_exception(error_);
-        if (stopping_ || bytes.empty() || bytes.size() > 1024 || bytes_ + bytes.size() > capacity ||
-            (paste_ && !tf_shell::paste_terminal(paste_->progress().state))) { ++rejected; return false; }
-        chunks_.push_back(bytes); bytes_ += bytes.size(); high_water = (std::max)(high_water, bytes_);
-        ready_.notify_one(); return true;
+        bool accepted = false;
+        {
+            std::lock_guard<std::mutex> lock(mutex_);
+            if (error_) std::rethrow_exception(error_);
+            if (stopping_ || bytes.empty() || bytes.size() > 1024 || bytes_ + bytes.size() > capacity ||
+                (paste_ && !tf_shell::paste_terminal(paste_->progress().state))) ++rejected;
+            else {
+                chunks_.push_back(bytes); bytes_ += bytes.size(); high_water = (std::max)(high_water, bytes_);
+                ready_.notify_one(); accepted = true;
+            }
+        }
+        if (trace_hook_) trace_hook_(this, false, bytes.size(), accepted);
+        return accepted;
     }
     // The queue is already bound to one Session. The UI retains the transfer
     // handle and target identity; later tab switches never retarget this queue.

experiments/shell/shell.cpp

@@ -51,6 +51,23 @@ void selection_discard_diagnostic(const termforge::TextSelection::DiscardReport&
            << ",\"current_bytes\":" << report.current_bytes << ",\"saved_columns\":" << report.saved_columns
            << ",\"current_columns\":" << report.current_columns << ",\"t_ms\":" << GetTickCount64() << "}\n";
 }
+// Each window owns its trace stream; the hook resolves it by queue pointer so a
+// window that is being destroyed can never leave a dangling global stream.
+struct TraceTarget { std::ofstream* stream = nullptr; uint64_t terminal = 0; };
+std::mutex g_trace_mutex;
+std::map<const void*, TraceTarget> g_trace_targets;
+void trace_input(const void* owner, bool written, size_t bytes, bool accepted) {
+    std::lock_guard<std::mutex> lock(g_trace_mutex);
+    const auto found = g_trace_targets.find(owner);
+    if (found == g_trace_targets.end() || !found->second.stream || !found->second.stream->good()) return;
+    std::ofstream& stream = *found->second.stream;
+    const uint64_t terminal = found->second.terminal;
+    stream << "{\"schema\":\"termforge-input-trace-v1\",\"terminal\":" << terminal
+                    << ",\"phase\":\"" << (written ? "write" : "enqueue")
+                    << "\",\"t_ms\":" << GetTickCount64() << ",\"bytes\":" << bytes
+                    << ",\"accepted\":" << (accepted ? "true" : "false") << "}\n";
+    stream.flush();
+}
 void install_selection_diagnostic_once() {
     static std::once_flag once;
     std::call_once(once, [] {
@@ -1020,7 +1037,13 @@ class ShellWindow {
         const std::string capture = env_path("TERMFORGE_CAPTURE_OUTPUT");
         if (!capture.empty()) capture_output_.open(capture, std::ios::binary | std::ios::app);
         const std::string trace = env_path("TERMFORGE_OUTPUT_TRACE");
-        if (!trace.empty()) trace_output_.open(trace, std::ios::app);
+        if (!trace.empty()) {
+            trace_output_.open(trace, std::ios::app);
+            if (trace_output_) {
+                { std::lock_guard<std::mutex> lock(g_trace_mutex); g_trace_targets[&input_] = TraceTarget{&trace_output_, identity_}; }
+                input_.set_trace_hook(&trace_input);
+            }
+        }
         snapshot_ = std::make_shared<const Snapshot>(viewport_.snapshot(parser_.screen), 0);
     }
     ~ShellWindow() {
@@ -1031,6 +1054,10 @@ class ShellWindow {
         output_.stop();
         if (closer_.joinable()) closer_.join();
         input_.stop();
+        // Unregister before this window's trace stream is destroyed so no later
+        // hook call can reach a closed stream.
+        { std::lock_guard<std::mutex> lock(g_trace_mutex); g_trace_targets.erase(&input_); }
+        input_.set_trace_hook(nullptr);
         if (hwnd_) DestroyWindow(hwnd_);
         if (registered_ && --class_users_ == 0) UnregisterClassW(L"TermForgeShell", GetModuleHandleW(nullptr));
     }