diff --git a/experiments/shell/queues.hpp b/experiments/shell/queues.hpp index 7d8e4480..c1cce824 100644 --- a/experiments/shell/queues.hpp +++ b/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 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 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. diff --git a/experiments/shell/shell.cpp b/experiments/shell/shell.cpp index fbc3e739..6a857d18 100644 --- a/experiments/shell/shell.cpp +++ b/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 g_trace_targets; +void trace_input(const void* owner, bool written, size_t bytes, bool accepted) { + std::lock_guard 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 lock(g_trace_mutex); g_trace_targets[&input_] = TraceTarget{&trace_output_, identity_}; } + input_.set_trace_hook(&trace_input); + } + } snapshot_ = std::make_shared(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 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)); }