66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
import sys
|
|
import os
|
|
import time
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
project_root = os.path.abspath(os.getcwd())
|
|
sys.path.append(project_root)
|
|
sys.path.append(os.path.join(project_root, 'src'))
|
|
|
|
from src import app_controller, gui_2, models, project_manager, events
|
|
|
|
def test_repro():
|
|
print("Initializing App...")
|
|
app = gui_2.App()
|
|
ctrl = app.controller
|
|
|
|
# 1. Scaffold a fake project
|
|
proj_path = os.path.abspath("tests/artifacts/repro_project.toml")
|
|
if os.path.exists(proj_path): os.remove(proj_path)
|
|
|
|
print(f"Scaffolding project at {proj_path}...")
|
|
proj = project_manager.default_project("Repro")
|
|
project_manager.save_project(proj, proj_path)
|
|
|
|
# 2. Load the project
|
|
ctrl.active_project_path = proj_path
|
|
ctrl.init_state()
|
|
|
|
# 3. Enable history
|
|
ctrl.ui_auto_add_history = True
|
|
print(f"ui_auto_add_history set to: {ctrl.ui_auto_add_history}")
|
|
|
|
# 4. Simulate a request event
|
|
print("Triggering user_request event...")
|
|
event_payload = events.UserRequestEvent(prompt="Hello Test", stable_md="Context", file_items=[], disc_text="", base_dir=".")
|
|
# The event is processed by _handle_request_event which calls ai_client.send
|
|
# But ai_client.send emits events via its global emitter.
|
|
|
|
from src import ai_client
|
|
# Simulate ai_client emitting the "request_start" event which should trigger _on_api_event
|
|
print("Emitting request_start event...")
|
|
ai_client.events.emit("request_start", payload={
|
|
"provider": "repro",
|
|
"model": "repro-model",
|
|
"message": "Hello Test"
|
|
}, kind="request")
|
|
|
|
# 5. Wait for history processor
|
|
print("Waiting for history processing...")
|
|
# Headless mode would call it in queue_fallback, but we can call it manually
|
|
time.sleep(0.5)
|
|
ctrl._process_pending_history_adds()
|
|
|
|
print(f"History entries: {len(ctrl.disc_entries)}")
|
|
for e in ctrl.disc_entries:
|
|
print(f" - {e.get('role')}: {e.get('content')}")
|
|
|
|
if len(ctrl.disc_entries) == 0:
|
|
print("REPRODUCED: History is empty!")
|
|
else:
|
|
print("SUCCESS: History updated.")
|
|
|
|
if __name__ == "__main__":
|
|
test_repro()
|