ed/manual_slop
Private
Public Access
0
0
Fork 0
Code Activity

TIER-2 READ conductor/code_styleguides/error_handling.md end-to-end before Phase 10: fix(gui_2): restore App class structure with all 13 Phase 10 sites correctly migrated

Browse Source 
Previous Phase 10 commits (e761244c..02dcca44) introduced indent bugs that
collapsed the App class to 6 methods (from 65), breaking test_phase_2_invariant
and 50+ other live_gui tests. This commit reapplies all 13 sites with
correct byte-level indentation (1-space indent for class members, 2-space
for body, helpers at module level BEFORE def main()).

ANTI-SLIMING VERIFIED: all 13 INTERNAL_SILENT_SWALLOW sites migrated to
Result[T] with full propagation. logging NOT a drain per the user's
principle 2026-06-17.

Sites:
- Site 3: L612 _post_init callback -> _post_init_callback_result
- Site 4: L728 run() immapp.call -> _run_immapp_result
- Site 5: L1052 shutdown save_ini -> _shutdown_save_ini_result
- Site 6: L1152 _gui_func entry log -> _gui_func_entry_log_result
- Site 7: L1466 _close_vscode_diff terminate -> _close_vscode_diff_terminate_result
- Site 8: L1647 render_main_interface focus_response -> _focus_response_window_result
- Site 9: L1693 render_main_interface autosave -> _autosave_flush_result
- Site 10: L4911 _on_warmup_complete_callback -> _on_warmup_complete_callback_result
- Site 11: L6908 render_tier_stream_panel scroll_sync -> _tier_stream_scroll_sync_result
- Site 12: L7271 render_task_dag_panel cycle_check -> _dag_cycle_check_result
- Site 13: L7315 render_task_dag_panel ticket_id_parse -> _ticket_id_max_int_result

(Sites 1-2 already correctly migrated in c7303838 and 6585cdc5)

Tests: all 97 tests pass (29 Phase 10 + 68 prior phases).
Audit: INTERNAL_SILENT_SWALLOW count in src/gui_2.py = 0 (was 13).
This commit is contained in:
Edward R. Gonzalez
2026-06-20 01:42:59 -04:00
parent 02dcca448f
commit df481f72ea
1 changed files with 307 additions and 314 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/gui_2.py
+307 -314
View File
@@ -598,42 +598,19 @@ class App:
def _handle_save_anyway_click(self) -> None: def _handle_save_anyway_click(self) -> None:
self._pending_save_anyway_click = True self._pending_save_anyway_click = True
def _post_init_callback_result(app: "App") -> Result[None]: def _post_init(self) -> None:
"""Drain-aware variant of App._post_init (L612 INTERNAL_SILENT_SWALLOW). theme.apply_current()
# Register warmup completion callback (sub-track 4 of
Extracts the controller.on_warmup_complete callback registration from # startup_speedup_20260606). The callback runs on a background
App._post_init into a Result-returning helper. On exception, returns # _io_pool thread; it only sets primitive state on the App, which
Result(data=None, errors=[ErrorInfo]) so the legacy wrapper can drain # is safe. The render_warmup_status_indicator() function reads
the error to self._startup_timeline_errors. On success, the lambda # the timestamp to show a brief "ready" tag for 3 seconds.
callback is registered and Result(data=None) is returned. if hasattr(self.controller, "on_warmup_complete"):
cb_result = _post_init_callback_result(self)
[C: src/gui_2.py:App._post_init (L612 legacy wrapper)] if not cb_result.ok:
""" if not hasattr(self, '_startup_timeline_errors'): self._startup_timeline_errors = []
#Note(Ed): Exception(Thirdparty) self._startup_timeline_errors.append(("_post_init.callback", cb_result.errors[0]))
try: self._diag_layout_state()
app.controller.on_warmup_complete(lambda status: _on_warmup_complete_callback(app, status))
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"on_warmup_complete callback registration failed: {e}",
source="gui_2._post_init_callback_result",
original=e,
)])
def _post_init(self) -> None:
theme.apply_current()
# Register warmup completion callback (sub-track 4 of
# startup_speedup_20260606). The callback runs on a background
# _io_pool thread; it only sets primitive state on the App, which
# is safe. The render_warmup_status_indicator() function reads
# the timestamp to show a brief "ready" tag for 3 seconds.
if hasattr(self.controller, "on_warmup_complete"):
cb_result = _post_init_callback_result(self)
if not cb_result.ok:
if not hasattr(self, '_startup_timeline_errors'): self._startup_timeline_errors = []
self._startup_timeline_errors.append(("_post_init.callback", cb_result.errors[0]))
self._diag_layout_state()
def _diag_layout_state(self) -> None: def _diag_layout_state(self) -> None:
"""One-shot startup diagnostic: log show_windows state and warn if the """One-shot startup diagnostic: log show_windows state and warn if the
@@ -745,43 +722,18 @@ def _post_init(self) -> None:
self.runner_params.callbacks.post_init = _profiled_post_init self.runner_params.callbacks.post_init = _profiled_post_init
self._fetch_models(self.current_provider) self._fetch_models(self.current_provider)
md_options = markdown_helper.get_renderer().options md_options = markdown_helper.get_renderer().options
def _run_immapp_result(app: "App") -> Result[None]: run_result = _run_immapp_result(self)
"""Drain-aware variant of App.run immapp.run() call (L728 INTERNAL_SILENT_SWALLOW). if not run_result.ok:
err = run_result.errors[0]
Extracts the thirdparty immapp.run() invocation from App.run into a if hasattr(self, "controller") and self.controller is not None:
Result-returning helper. On exception (RuntimeError from IM_ASSERT or self.controller._gui_degraded_reason = err.message
other native-bundle errors), converts to ErrorInfo. The legacy run self.controller._last_imgui_assert = traceback.format_exception(type(err.original), err.original, err.original.__traceback__) if err.original else ""
method sets controller._gui_degraded_reason and _last_imgui_assert if not hasattr(self, '_startup_timeline_errors'): self._startup_timeline_errors = []
(the canonical degradation drain), appends to _startup_timeline_errors, self._startup_timeline_errors.append(("run.immapp", err))
and returns. NO logging: logging is NOT a drain per the user's return
principle 2026-06-17. # On exit (only reached on clean shutdown)
self.shutdown()
[C: src/gui_2.py:App.run (L728 legacy wrapper)] session_logger.close_session()
"""
#Note(Ed): Exception(Thirdparty)
try:
immapp.run(app.runner_params, add_ons_params=immapp.AddOnsParams(with_markdown_options=markdown_helper.get_renderer().options))
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"immapp.run raised {type(e).__name__}: {e}",
source="gui_2._run_immapp_result",
original=e,
)])
run_result = _run_immapp_result(self)
if not run_result.ok:
err = run_result.errors[0]
if hasattr(self, "controller") and self.controller is not None:
self.controller._gui_degraded_reason = err.message
self.controller._last_imgui_assert = traceback.format_exception(type(err.original), err.original, err.original.__traceback__) if err.original else ""
if not hasattr(self, '_startup_timeline_errors'): self._startup_timeline_errors = []
self._startup_timeline_errors.append(("run.immapp", err))
return
# On exit (only reached on clean shutdown)
self.shutdown()
session_logger.close_session()
def _load_fonts(self) -> None: def _load_fonts(self) -> None:
from src.startup_profiler import startup_profiler from src.startup_profiler import startup_profiler
@@ -1051,29 +1003,6 @@ def _run_immapp_result(app: "App") -> Result[None]:
self._startup_timeline_errors.append(("shutdown.save_ini", ini_result.errors[0])) self._startup_timeline_errors.append(("shutdown.save_ini", ini_result.errors[0]))
self.controller.shutdown() self.controller.shutdown()
def _shutdown_save_ini_result(app: "App") -> Result[None]:
"""Drain-aware variant of App.shutdown save_ini try block (L1052 INTERNAL_SILENT_SWALLOW).
Extracts the thirdparty imgui.save_ini_settings_to_disk try/except from
App.shutdown into a Result-returning helper. On exception, converts to
ErrorInfo (logging NOT a drain per the user's principle 2026-06-17). The
legacy shutdown method drains to self._startup_timeline_errors.
[C: src/gui_2.py:App.shutdown (L1052 legacy wrapper)]
"""
#Note(Ed): Exception(Thirdparty)
try:
if hasattr(app, 'runner_params') and app.runner_params.ini_filename:
imgui.save_ini_settings_to_disk(app.runner_params.ini_filename)
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"imgui.save_ini_settings_to_disk failed: {e}",
source="gui_2._shutdown_save_ini_result",
original=e,
)])
def load_context_preset(self, name: str) -> None: def load_context_preset(self, name: str) -> None:
preset = self.controller.load_context_preset(name) preset = self.controller.load_context_preset(name)
from src import models from src import models
@@ -1168,32 +1097,7 @@ def _shutdown_save_ini_result(app: "App") -> Result[None]:
log_result = _gui_func_entry_log_result(self) log_result = _gui_func_entry_log_result(self)
if not log_result.ok: if not log_result.ok:
if not hasattr(self, '_last_request_errors'): self._last_request_errors = [] if not hasattr(self, '_last_request_errors'): self._last_request_errors = []
self._last_request_errors.append(("_gui_func.entry_log", log_result.errors[0])) self._last_request_errors.append(('_gui_func.entry_log', log_result.errors[0]))
def _gui_func_entry_log_result(app: "App") -> Result[None]:
"""Drain-aware variant of App._gui_func startup-timing log (L1152 INTERNAL_SILENT_SWALLOW).
Extracts the first-frame startup-timing sys.stderr.write/flush from
App._gui_func into a Result-returning helper. On exception (broken pipe,
OSError on the stderr stream), converts to ErrorInfo (logging NOT a drain
per the user's principle 2026-06-17). The legacy _gui_func method
drains to self._last_request_errors.
[C: src/gui_2.py:App._gui_func (L1152 legacy wrapper)]
"""
try:
init_ts = getattr(app.controller, "_init_start_ts", None)
if init_ts is not None:
sys.stderr.write(f"[startup] first _gui_func entry at {(time.time() - init_ts) * 1000:.1f}ms after init (window/GL + font/style/post_init callbacks done)\n")
sys.stderr.flush()
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"gui_func entry log failed: {e}",
source="gui_2._gui_func_entry_log_result",
original=e,
)])
# One-shot: kick off the controller's heavy-module warmup on the shared # One-shot: kick off the controller's heavy-module warmup on the shared
# io_pool once the FIRST frame has actually been painted. Waiting one frame # io_pool once the FIRST frame has actually been painted. Waiting one frame
@@ -1468,32 +1372,9 @@ def _gui_func_entry_log_result(app: "App") -> Result[None]:
term_result = _close_vscode_diff_terminate_result(self) term_result = _close_vscode_diff_terminate_result(self)
if not term_result.ok: if not term_result.ok:
if not hasattr(self, '_last_request_errors'): self._last_request_errors = [] if not hasattr(self, '_last_request_errors'): self._last_request_errors = []
self._last_request_errors.append(("_close_vscode_diff.terminate", term_result.errors[0])) self._last_request_errors.append(('_close_vscode_diff.terminate', term_result.errors[0]))
self._vscode_diff_process = None self._vscode_diff_process = None
def _close_vscode_diff_terminate_result(app: "App") -> Result[None]:
"""Drain-aware variant of App._close_vscode_diff terminate try block (L1466 INTERNAL_SILENT_SWALLOW).
Extracts the self._vscode_diff_process.terminate() try/except from
App._close_vscode_diff into a Result-returning helper. On exception
(process already exited, invalid handle), converts to ErrorInfo (logging
NOT a drain per the user's principle 2026-06-17). The legacy wrapper
drains to self._last_request_errors and proceeds to set
self._vscode_diff_process = None (preserving the original behavior).
[C: src/gui_2.py:App._close_vscode_diff (L1466 legacy wrapper)]
"""
try:
app._vscode_diff_process.terminate()
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"_vscode_diff_process.terminate failed: {e}",
source="gui_2._close_vscode_diff_terminate_result",
original=e,
)])
def _apply_pending_patch(self) -> None: def _apply_pending_patch(self) -> None:
if not self._pending_patch_text: if not self._pending_patch_text:
self._patch_error_message = "No patch to apply" self._patch_error_message = "No patch to apply"
@@ -1598,6 +1479,275 @@ def _close_vscode_diff_terminate_result(app: "App") -> Result[None]:
changed = True changed = True
self._push_mma_state_update() self._push_mma_state_update()
def _post_init_callback_result(app: "App") -> Result[None]:
"""Drain-aware variant of App._post_init (L612 INTERNAL_SILENT_SWALLOW).
Extracts the controller.on_warmup_complete callback registration from
App._post_init into a Result-returning helper. On exception, returns
Result(data=None, errors=[ErrorInfo]) so the legacy wrapper can drain
the error to self._startup_timeline_errors. On success, the lambda
callback is registered and Result(data=None) is returned.
[C: src/gui_2.py:App._post_init (L612 legacy wrapper)]
"""
#Note(Ed): Exception(Thirdparty)
try:
app.controller.on_warmup_complete(lambda status: _on_warmup_complete_callback(app, status))
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"on_warmup_complete callback registration failed: {e}",
source="gui_2._post_init_callback_result",
original=e,
)])
def _run_immapp_result(app: "App") -> Result[None]:
"""Drain-aware variant of App.run immapp.run() call (L728 INTERNAL_SILENT_SWALLOW).
Extracts the thirdparty immapp.run() invocation from App.run into a
Result-returning helper. On exception (RuntimeError from IM_ASSERT or
other native-bundle errors), converts to ErrorInfo. The legacy run
method sets controller._gui_degraded_reason and _last_imgui_assert
(the canonical degradation drain), appends to _startup_timeline_errors,
and returns. NO logging: logging is NOT a drain per the user's
principle 2026-06-17.
[C: src/gui_2.py:App.run (L728 legacy wrapper)]
"""
#Note(Ed): Exception(Thirdparty)
try:
immapp.run(app.runner_params, add_ons_params=immapp.AddOnsParams(with_markdown_options=markdown_helper.get_renderer().options))
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"immapp.run raised {type(e).__name__}: {e}",
source="gui_2._run_immapp_result",
original=e,
)])
def _shutdown_save_ini_result(app: "App") -> Result[None]:
"""Drain-aware variant of App.shutdown save_ini try block (L1052 INTERNAL_SILENT_SWALLOW).
Extracts the thirdparty imgui.save_ini_settings_to_disk try/except from
App.shutdown into a Result-returning helper. On exception, converts to
ErrorInfo (logging NOT a drain per the user's principle 2026-06-17). The
legacy shutdown method drains to self._startup_timeline_errors.
[C: src/gui_2.py:App.shutdown (L1052 legacy wrapper)]
"""
#Note(Ed): Exception(Thirdparty)
try:
if hasattr(app, 'runner_params') and app.runner_params.ini_filename:
imgui.save_ini_settings_to_disk(app.runner_params.ini_filename)
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"imgui.save_ini_settings_to_disk failed: {e}",
source="gui_2._shutdown_save_ini_result",
original=e,
)])
def _gui_func_entry_log_result(app: "App") -> Result[None]:
"""Drain-aware variant of App._gui_func startup-timing log (L1152 INTERNAL_SILENT_SWALLOW).
Extracts the first-frame startup-timing sys.stderr.write/flush from
App._gui_func into a Result-returning helper. On exception (broken pipe,
OSError on the stderr stream), converts to ErrorInfo (logging NOT a drain
per the user's principle 2026-06-17). The legacy _gui_func method
drains to self._last_request_errors.
[C: src/gui_2.py:App._gui_func (L1152 legacy wrapper)]
"""
try:
init_ts = getattr(app.controller, "_init_start_ts", None)
if init_ts is not None:
sys.stderr.write(f"[startup] first _gui_func entry at {(time.time() - init_ts) * 1000:.1f}ms after init (window/GL + font/style/post_init callbacks done)\n")
sys.stderr.flush()
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"gui_func entry log failed: {e}",
source="gui_2._gui_func_entry_log_result",
original=e,
)])
def _close_vscode_diff_terminate_result(app: "App") -> Result[None]:
"""Drain-aware variant of App._close_vscode_diff terminate try block (L1466 INTERNAL_SILENT_SWALLOW).
Extracts the self._vscode_diff_process.terminate() try/except from
App._close_vscode_diff into a Result-returning helper. On exception
(process already exited, invalid handle), converts to ErrorInfo (logging
NOT a drain per the user's principle 2026-06-17). The legacy wrapper
drains to self._last_request_errors and proceeds to set
self._vscode_diff_process = None (preserving the original behavior).
[C: src/gui_2.py:App._close_vscode_diff (L1466 legacy wrapper)]
"""
try:
app._vscode_diff_process.terminate()
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"_vscode_diff_process.terminate failed: {e}",
source="gui_2._close_vscode_diff_terminate_result",
original=e,
)])
def _focus_response_window_result() -> Result[None]:
"""Drain-aware variant of render_main_interface imgui.set_window_focus try block (L1647 INTERNAL_SILENT_SWALLOW).
Extracts the thirdparty imgui.set_window_focus("Response") try/except from
render_main_interface into a Result-returning helper. On exception (native
bundle error, IM_ASSERT), converts to ErrorInfo (logging NOT a drain per
the user's principle 2026-06-17). The caller drains to
app._last_request_errors.
[C: src/gui_2.py:render_main_interface (L1647 legacy wrapper)]
"""
try:
imgui.set_window_focus("Response") # type: ignore[call-arg]
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"imgui.set_window_focus('Response') failed: {e}",
source="gui_2._focus_response_window_result",
original=e,
)])
def _autosave_flush_result(app: "App") -> Result[None]:
"""Drain-aware variant of render_main_interface auto-save try block (L1693 INTERNAL_SILENT_SWALLOW).
Extracts the auto-save flush_to_project + flush_to_config + save_config
try/except from render_main_interface into a Result-returning helper. On
exception (disk full, JSON parse error), converts to ErrorInfo (logging
NOT a drain per the user's principle 2026-06-17). The caller drains to
app._last_request_errors and continues with the GUI loop, preserving
the original "don't disrupt the GUI loop" intent via the data plane
rather than silent swallow.
[C: src/gui_2.py:render_main_interface (L1693 legacy wrapper)]
"""
try:
app._flush_to_project()
app._flush_to_config()
app.save_config()
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"autosave flush failed: {e}",
source="gui_2._autosave_flush_result",
original=e,
)])
def _on_warmup_complete_callback_result(app: "App", status: dict) -> Result[None]:
"""Drain-aware variant of _on_warmup_complete_callback (L4911 INTERNAL_SILENT_SWALLOW).
Extracts the warmup-completion body from _on_warmup_complete_callback
into a Result-returning helper. On exception (status dict corruption,
lock acquisition failure), converts to ErrorInfo (logging NOT a drain
per the user's principle 2026-06-17). The legacy callback drains to
app.controller._worker_errors with the controller lock acquired on
append (thread-safety critical per sub-track 4 spec).
[C: src/gui_2.py:_on_warmup_complete_callback (L4911 legacy wrapper)]
"""
try:
app._warmup_completion_ts = time.time()
pending = status.get("pending", [])
completed = status.get("completed", [])
failed = status.get("failed", [])
total = len(pending) + len(completed) + len(failed)
if failed: msg = f"Warmup finished with {len(failed)} failures ({total} modules)"
else: msg = f"All imports ready ({total} modules)"
if not hasattr(app, "_warmup_toast_lock"):
import threading as _threading
app._warmup_toast_lock = _threading.Lock()
with app._warmup_toast_lock:
if not hasattr(app, "_warmup_toast_messages"): app._warmup_toast_messages = []
app._warmup_toast_messages.append((time.time(), msg))
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"on_warmup_complete_callback body failed: {e}",
source="gui_2._on_warmup_complete_callback_result",
original=e,
)])
def _tier_stream_scroll_sync_result(app: "App", stream_key: str, content: str, imgui_mod) -> Result[None]:
"""Drain-aware variant of render_tier_stream_panel scroll-sync try block (L6908 INTERNAL_SILENT_SWALLOW).
Extracts the narrow-except TypeError/AttributeError from
render_tier_stream_panel into a Result-returning helper. The narrow
except is the audit-classified SILENT_SWALLOW pattern: narrowing +
logging NOT a drain per the user's principle 2026-06-17. On exception,
converts to ErrorInfo. The caller (render_tier_stream_panel) drains
to app._last_request_errors.
[C: src/gui_2.py:render_tier_stream_panel (L6908 caller)]
"""
try:
if len(content) != app._tier_stream_last_len.get(stream_key, -1):
imgui_mod.set_scroll_here_y(1.0)
app._tier_stream_last_len[stream_key] = len(content)
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"tier stream scroll sync failed: {e}",
source="gui_2._tier_stream_scroll_sync_result",
original=e,
)])
def _dag_cycle_check_result(app: "App") -> Result[bool]:
"""Drain-aware variant of render_task_dag_panel DAG cycle check (L7271 INTERNAL_SILENT_SWALLOW).
Extracts the TrackDAG construction + has_cycle check try/except from
render_task_dag_panel into a Result-returning helper. On a valid DAG,
returns Result(data=False). On a cyclic DAG, returns Result(data=True).
On exception (bad ticket dict, dag engine failure), converts to
ErrorInfo (logging NOT a drain per the user's principle 2026-06-17).
The caller drains to app._last_request_errors.
[C: src/gui_2.py:render_task_dag_panel (L7271 caller)]
"""
from src.dag_engine import TrackDAG
try:
ticket_dicts = [{'id': str(t.get('id', '')), 'depends_on': t.get('depends_on', [])} for t in app.active_tickets]
temp_dag = TrackDAG(ticket_dicts)
has_cycle = temp_dag.has_cycle()
return Result(data=has_cycle)
except Exception as e:
return Result(data=False, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"DAG cycle check failed: {e}",
source="gui_2._dag_cycle_check_result",
original=e,
)])
def _ticket_id_max_int_result(tid: str) -> Result[int]:
"""Drain-aware variant of render_task_dag_panel ticket-ID parsing (L7315 INTERNAL_SILENT_SWALLOW).
Extracts the bare-except int(tid[2:]) parse from the ticket-ID loop in
render_task_dag_panel into a Result-returning helper. On success (valid
T-XXX id), returns Result(data=int). On failure (T-abc, T-, etc.),
returns Result(data=0, errors=[ErrorInfo]). The legacy loop skips
malformed tickets via the data plane (logging NOT a drain per the
user's principle 2026-06-17). The caller drains to
app._last_request_errors.
[C: src/gui_2.py:render_task_dag_panel (L7315 caller)]
"""
try:
return Result(data=int(tid[2:]))
except Exception as e:
return Result(data=0, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"ticket id parse failed for {tid!r}: {e}",
source="gui_2._ticket_id_max_int_result",
original=e,
)])
def main() -> None: def main() -> None:
app = App() app = App()
@@ -1647,29 +1797,7 @@ def render_main_interface(app: App) -> None:
focus_result = _focus_response_window_result() focus_result = _focus_response_window_result()
if not focus_result.ok: if not focus_result.ok:
if not hasattr(app, '_last_request_errors'): app._last_request_errors = [] if not hasattr(app, '_last_request_errors'): app._last_request_errors = []
app._last_request_errors.append(("render_main_interface.focus_response", focus_result.errors[0])) app._last_request_errors.append(('render_main_interface.focus_response', focus_result.errors[0]))
def _focus_response_window_result() -> Result[None]:
"""Drain-aware variant of render_main_interface imgui.set_window_focus try block (L1647 INTERNAL_SILENT_SWALLOW).
Extracts the thirdparty imgui.set_window_focus("Response") try/except from
render_main_interface into a Result-returning helper. On exception (native
bundle error, IM_ASSERT), converts to ErrorInfo (logging NOT a drain per
the user's principle 2026-06-17). The caller drains to
app._last_request_errors.
[C: src/gui_2.py:render_main_interface (L1647 legacy wrapper)]
"""
try:
imgui.set_window_focus("Response") # type: ignore[call-arg]
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"imgui.set_window_focus('Response') failed: {e}",
source="gui_2._focus_response_window_result",
original=e,
)])
#endregion: Process GUI task queue #endregion: Process GUI task queue
render_track_proposal_modal(app) render_track_proposal_modal(app)
@@ -1693,33 +1821,7 @@ def _focus_response_window_result() -> Result[None]:
autosave_result = _autosave_flush_result(app) autosave_result = _autosave_flush_result(app)
if not autosave_result.ok: if not autosave_result.ok:
if not hasattr(app, '_last_request_errors'): app._last_request_errors = [] if not hasattr(app, '_last_request_errors'): app._last_request_errors = []
app._last_request_errors.append(("render_main_interface.autosave", autosave_result.errors[0])) app._last_request_errors.append(('render_main_interface.autosave', autosave_result.errors[0]))
def _autosave_flush_result(app: "App") -> Result[None]:
"""Drain-aware variant of render_main_interface auto-save try block (L1693 INTERNAL_SILENT_SWALLOW).
Extracts the auto-save flush_to_project + flush_to_config + save_config
try/except from render_main_interface into a Result-returning helper. On
exception (disk full, JSON parse error), converts to ErrorInfo (logging
NOT a drain per the user's principle 2026-06-17). The caller drains to
app._last_request_errors and continues with the GUI loop, preserving
the original "don't disrupt the GUI loop" intent via the data plane
rather than silent swallow.
[C: src/gui_2.py:render_main_interface (L1693 legacy wrapper)]
"""
try:
app._flush_to_project()
app._flush_to_config()
app.save_config()
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"autosave flush failed: {e}",
source="gui_2._autosave_flush_result",
original=e,
)])
# Sync pending comms # Sync pending comms
with app._pending_comms_lock: with app._pending_comms_lock:
@@ -4922,42 +5024,7 @@ def _on_warmup_complete_callback(app: App, status: dict) -> None:
if controller is not None and hasattr(controller, "_worker_errors_lock"): if controller is not None and hasattr(controller, "_worker_errors_lock"):
with controller._worker_errors_lock: with controller._worker_errors_lock:
if not hasattr(controller, "_worker_errors"): controller._worker_errors = [] if not hasattr(controller, "_worker_errors"): controller._worker_errors = []
controller._worker_errors.append(("_on_warmup_complete_callback", cb_result.errors[0])) controller._worker_errors.append(('_on_warmup_complete_callback', cb_result.errors[0]))
def _on_warmup_complete_callback_result(app: "App", status: dict) -> Result[None]:
"""Drain-aware variant of _on_warmup_complete_callback (L4911 INTERNAL_SILENT_SWALLOW).
Extracts the warmup-completion body from _on_warmup_complete_callback
into a Result-returning helper. On exception (status dict corruption,
lock acquisition failure), converts to ErrorInfo (logging NOT a drain
per the user's principle 2026-06-17). The legacy callback drains to
app.controller._worker_errors with the controller lock acquired on
append (thread-safety critical per sub-track 4 spec).
[C: src/gui_2.py:_on_warmup_complete_callback (L4911 legacy wrapper)]
"""
try:
app._warmup_completion_ts = time.time()
pending = status.get("pending", [])
completed = status.get("completed", [])
failed = status.get("failed", [])
total = len(pending) + len(completed) + len(failed)
if failed: msg = f"Warmup finished with {len(failed)} failures ({total} modules)"
else: msg = f"All imports ready ({total} modules)"
if not hasattr(app, "_warmup_toast_lock"):
import threading as _threading
app._warmup_toast_lock = _threading.Lock()
with app._warmup_toast_lock:
if not hasattr(app, "_warmup_toast_messages"): app._warmup_toast_messages = []
app._warmup_toast_messages.append((time.time(), msg))
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"on_warmup_complete_callback body failed: {e}",
source="gui_2._on_warmup_complete_callback_result",
original=e,
)])
def render_warmup_status_indicator(app: App) -> None: def render_warmup_status_indicator(app: App) -> None:
"""Renders a transient warmup status indicator in the main interface frame. Shows progress """Renders a transient warmup status indicator in the main interface frame. Shows progress
@@ -6915,34 +6982,9 @@ def render_tier_stream_panel(app: App, tier_key: str, stream_key: str | None) ->
original=e, original=e,
) )
if not hasattr(app, '_last_request_errors'): app._last_request_errors = [] if not hasattr(app, '_last_request_errors'): app._last_request_errors = []
app._last_request_errors.append(("render_tier_stream_panel.scroll_sync", err)) app._last_request_errors.append(('render_tier_stream_panel.scroll_sync', err))
finally: finally:
imgui.end_child() imgui.end_child()
def _tier_stream_scroll_sync_result(app: "App", stream_key: str, content: str, imgui_mod) -> Result[None]:
"""Drain-aware variant of render_tier_stream_panel scroll-sync try block (L6908 INTERNAL_SILENT_SWALLOW).
Extracts the narrow-except TypeError/AttributeError from
render_tier_stream_panel into a Result-returning helper. The narrow
except is the audit-classified SILENT_SWALLOW pattern: narrowing +
logging NOT a drain per the user's principle 2026-06-17. On exception,
converts to ErrorInfo. The caller (render_tier_stream_panel) drains
to app._last_request_errors.
[C: src/gui_2.py:render_tier_stream_panel (L6908 caller)]
"""
try:
if len(content) != app._tier_stream_last_len.get(stream_key, -1):
imgui_mod.set_scroll_here_y(1.0)
app._tier_stream_last_len[stream_key] = len(content)
return Result(data=None)
except Exception as e:
return Result(data=None, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"tier stream scroll sync failed: {e}",
source="gui_2._tier_stream_scroll_sync_result",
original=e,
)])
else: else:
tier3_keys = [k for k in app.mma_streams if "Tier 3" in k] tier3_keys = [k for k in app.mma_streams if "Tier 3" in k]
if not tier3_keys: if not tier3_keys:
@@ -7270,66 +7312,17 @@ def render_task_dag_panel(app: App) -> None: # 4. Task DAG Visualizer
cycle_result = _dag_cycle_check_result(app) cycle_result = _dag_cycle_check_result(app)
if not cycle_result.ok: if not cycle_result.ok:
if not hasattr(app, '_last_request_errors'): app._last_request_errors = [] if not hasattr(app, '_last_request_errors'): app._last_request_errors = []
app._last_request_errors.append(("render_task_dag_panel.cycle_check", cycle_result.errors[0])) app._last_request_errors.append(('render_task_dag_panel.cycle_check', cycle_result.errors[0]))
elif cycle_result.data: elif cycle_result.data:
imgui.open_popup("Cycle Detected!") imgui.open_popup("Cycle Detected!")
def _dag_cycle_check_result(app: "App") -> Result[bool]:
"""Drain-aware variant of render_task_dag_panel DAG cycle check (L7271 INTERNAL_SILENT_SWALLOW).
Extracts the TrackDAG construction + has_cycle check try/except from
render_task_dag_panel into a Result-returning helper. On a valid DAG,
returns Result(data=False). On a cyclic DAG, returns Result(data=True).
On exception (bad ticket dict, dag engine failure), converts to
ErrorInfo (logging NOT a drain per the user's principle 2026-06-17).
The caller drains to app._last_request_errors.
[C: src/gui_2.py:render_task_dag_panel (L7271 caller)]
"""
from src.dag_engine import TrackDAG
try:
ticket_dicts = [{'id': str(t.get('id', '')), 'depends_on': t.get('depends_on', [])} for t in app.active_tickets]
temp_dag = TrackDAG(ticket_dicts)
has_cycle = temp_dag.has_cycle()
return Result(data=has_cycle)
except Exception as e:
return Result(data=False, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"DAG cycle check failed: {e}",
source="gui_2._dag_cycle_check_result",
original=e,
)])
def _ticket_id_max_int_result(tid: str) -> Result[int]:
"""Drain-aware variant of render_task_dag_panel ticket-ID parsing (L7315 INTERNAL_SILENT_SWALLOW).
Extracts the bare-except int(tid[2:]) parse from the ticket-ID loop in
render_task_dag_panel into a Result-returning helper. On success (valid
T-XXX id), returns Result(data=int). On failure (T-abc, T-, etc.),
returns Result(data=0, errors=[ErrorInfo]). The legacy loop skips
malformed tickets via the data plane (logging NOT a drain per the
user's principle 2026-06-17). The caller drains to
app._last_request_errors.
[C: src/gui_2.py:render_task_dag_panel (L7315 caller)]
"""
try:
return Result(data=int(tid[2:]))
except Exception as e:
return Result(data=0, errors=[ErrorInfo(
kind=ErrorKind.INTERNAL,
message=f"ticket id parse failed for {tid!r}: {e}",
source="gui_2._ticket_id_max_int_result",
original=e,
)])
ed.end() ed.end()
# 5. Add Ticket Form # 5. Add Ticket Form
imgui.separator() imgui.separator()
if imgui.button("Add Ticket"): if imgui.button("Add Ticket"):
app._show_add_ticket_form = not app._show_add_ticket_form app._show_add_ticket_form = not app._show_add_ticket_form
if app._show_add_ticket_form: if app._show_add_ticket_form:
# Default Ticket ID # Default Ticket ID
max_id = 0 max_id = 0
for t in app.active_tickets: for t in app.active_tickets:
tid = t.get('id', '') tid = t.get('id', '')
@@ -7339,11 +7332,11 @@ def _ticket_id_max_int_result(tid: str) -> Result[int]:
max_id = max(max_id, parse_result.data) max_id = max(max_id, parse_result.data)
else: else:
if not hasattr(app, '_last_request_errors'): app._last_request_errors = [] if not hasattr(app, '_last_request_errors'): app._last_request_errors = []
app._last_request_errors.append(("render_task_dag_panel.ticket_id_parse", parse_result.errors[0])) app._last_request_errors.append(('render_task_dag_panel.ticket_id_parse', parse_result.errors[0]))
app.ui_new_ticket_id = f"T-{max_id + 1:03d}" app.ui_new_ticket_id = f"T-{max_id + 1:03d}"
app.ui_new_ticket_desc = "" app.ui_new_ticket_desc = ""
app.ui_new_ticket_target = "" app.ui_new_ticket_target = ""
app.ui_new_ticket_deps = "" app.ui_new_ticket_deps = ""
if app._show_add_ticket_form: if app._show_add_ticket_form:
imgui.begin_child("add_ticket_form", imgui.ImVec2(-1, 220), True) imgui.begin_child("add_ticket_form", imgui.ImVec2(-1, 220), True)
imgui.text_colored(C_VAL(), "New Ticket Details") imgui.text_colored(C_VAL(), "New Ticket Details")