-
Notifications
You must be signed in to change notification settings - Fork 115
Expand file tree
/
Copy pathcore.py
More file actions
3513 lines (3113 loc) · 152 KB
/
Copy pathcore.py
File metadata and controls
3513 lines (3113 loc) · 152 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""Interactive REPL for Claw Codex."""
from __future__ import annotations
try:
from prompt_toolkit import PromptSession
from prompt_toolkit.history import FileHistory
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.styles import Style
from prompt_toolkit.completion import Completer, Completion, WordCompleter
from prompt_toolkit.input import ansi_escape_sequences as _pt_ansi_seq
from prompt_toolkit.keys import Keys as _PTKeys
# Teach prompt_toolkit to distinguish Shift+Enter from plain Enter.
#
# Two distinct Shift+Enter sequences are in the wild; we route both to
# the same two-key tuple Meta+Enter uses (Escape + ControlM), so a
# single ``escape, c-m`` binding covers them all:
#
# 1. ``\x1b[13;2u`` — Kitty keyboard protocol (Kitty, WezTerm, Ghostty,
# iTerm2 with CSI u mode). Not known to prompt_toolkit at all.
# 2. ``\x1b[27;2;13~`` — xterm ``modifyOtherKeys`` level 2 (xterm with
# modifyOtherKeys on, some VSCode configurations). prompt_toolkit
# maps this to plain ``ControlM``, so by default it's
# indistinguishable from Enter — we override it.
#
# This matches the TypeScript reference's behavior in ``useTextInput.ts``
# which explicitly treats both CSI 13;2u and CSI 27;2;13~ as "insert
# newline" on Shift+Enter.
if not hasattr(_pt_ansi_seq, "_clawcodex_shift_enter_registered"):
_pt_ansi_seq.ANSI_SEQUENCES["\x1b[13;2u"] = (
_PTKeys.Escape,
_PTKeys.ControlM,
)
_pt_ansi_seq.ANSI_SEQUENCES["\x1b[27;2;13~"] = (
_PTKeys.Escape,
_PTKeys.ControlM,
)
_pt_ansi_seq._clawcodex_shift_enter_registered = True # type: ignore[attr-defined]
try:
from prompt_toolkit.completion import FuzzyCompleter
except Exception: # pragma: no cover
FuzzyCompleter = None # type: ignore
from prompt_toolkit.key_binding import KeyBindings
_HAS_PROMPT_TOOLKIT = True
except ModuleNotFoundError: # pragma: no cover
_HAS_PROMPT_TOOLKIT = False
class FileHistory: # type: ignore
def __init__(self, *args, **kwargs):
pass
class AutoSuggestFromHistory: # type: ignore
def __init__(self, *args, **kwargs):
pass
class Style: # type: ignore
@staticmethod
def from_dict(*args, **kwargs):
return None
class WordCompleter: # type: ignore
def __init__(self, *args, **kwargs):
pass
class Completer: # type: ignore
def get_completions(self, *args, **kwargs):
return iter(())
class Completion: # type: ignore
def __init__(self, *args, **kwargs):
pass
FuzzyCompleter = None # type: ignore
class KeyBindings: # type: ignore
def __init__(self, *args, **kwargs):
pass
class PromptSession: # type: ignore
def __init__(self, *args, **kwargs):
pass
def prompt(self, *args, **kwargs):
raise EOFError()
def _fuzzy_subseq(name: str, partial: str) -> bool:
"""Lightweight subsequence match (``partial`` chars appear in order)."""
if not partial:
return True
i = 0
for ch in name:
if ch == partial[i]:
i += 1
if i == len(partial):
return True
return False
class _SlashOnlyCompleter(Completer):
"""Trigger autocompletion only for slash commands, matching the reference
Claude Code behavior.
Rules (mirrors ``typescript/src/utils/suggestions/commandSuggestions.ts``):
* If the whole buffer starts with ``/`` and the cursor is on the first
token, complete slash commands (prefix match against the command name).
* If the cursor sits on a ``/``-prefixed token preceded by whitespace,
complete that mid-input slash command.
* In every other case (plain words like ``hello``, ``ex``, etc.) return
no completions so the user can type freely without a suggestion popup.
When a ``suggestions_provider`` is supplied it carries descriptions and
optional ``[workflow]`` tags, which surface in the prompt_toolkit menu
as ``display_meta`` — the same two-column layout the TS reference uses.
The legacy ``words_provider`` is still honoured for callers that only
have the flat name list.
"""
def __init__(self, words_provider, suggestions_provider=None):
self._words_provider = words_provider
self._suggestions_provider = suggestions_provider
def get_completions(self, document, complete_event): # type: ignore[override]
text = document.text_before_cursor
token, token_start = self._current_slash_token(text)
if token is None:
return
partial = token[1:].lower() # strip leading '/'
start_position = token_start - len(text)
if self._suggestions_provider is not None:
try:
suggestions = self._suggestions_provider() or []
except Exception:
suggestions = []
yield from self._rich_completions(suggestions, partial, start_position)
return
words = self._words_provider() or []
seen: set[str] = set()
for word in words:
if not isinstance(word, str) or not word.startswith("/"):
continue
name = word[1:]
key = name.lower()
if key in seen:
continue
if not partial or key.startswith(partial):
seen.add(key)
yield Completion(
text=word,
start_position=start_position,
display=word,
)
def _rich_completions(self, suggestions, partial, start_position):
"""Yield ``Completion`` entries with ``display`` + ``display_meta``.
Matches the TS ranking: exact name → exact alias → prefix name →
prefix alias → fuzzy. Aliases are surfaced in ``(alias)`` only
when the typed prefix matched the alias, so an unmatched partial
does not pollute the menu with every alternate name.
"""
scored: list[tuple[int, int, Any, str | None]] = []
seen: set[str] = set()
for idx, sugg in enumerate(suggestions):
name = getattr(sugg, "name", None)
if not isinstance(name, str) or not name:
continue
name_lc = name.lower()
if name_lc in seen:
continue
aliases = tuple(getattr(sugg, "aliases", ()) or ())
matched_alias: str | None = None
rank: int | None = None
if not partial:
rank = 0
elif name_lc == partial:
rank = 0
else:
exact_alias = next(
(a for a in aliases if a.lower() == partial), None
)
if exact_alias:
rank = 1
matched_alias = exact_alias
elif name_lc.startswith(partial):
rank = 2
else:
prefix_alias = next(
(a for a in aliases if a.lower().startswith(partial)),
None,
)
if prefix_alias:
rank = 3
matched_alias = prefix_alias
elif _fuzzy_subseq(name_lc, partial):
rank = 5
else:
fuzzy_alias = next(
(a for a in aliases if _fuzzy_subseq(a.lower(), partial)),
None,
)
if fuzzy_alias:
rank = 6
matched_alias = fuzzy_alias
if rank is None:
continue
seen.add(name_lc)
secondary = idx if not partial else len(name)
scored.append((rank, secondary, sugg, matched_alias))
scored.sort(key=lambda t: (t[0], t[1], t[2].name.lower()))
for _, _, sugg, matched_alias in scored:
alias_text = f" ({matched_alias})" if matched_alias else ""
display_text = f"/{sugg.name}{alias_text}"
display_styled = [("class:completion.command", display_text)]
description = (getattr(sugg, "description", "") or "").strip()
tag = getattr(sugg, "tag", None)
meta_parts: list[tuple[str, str]] = []
if tag:
meta_parts.append(("class:completion.tag", f"[{tag}] "))
if description:
# Collapse internal whitespace so multi-line descriptions
# render as one row in the prompt_toolkit menu.
meta_parts.append(
("class:completion.description", " ".join(description.split()))
)
yield Completion(
text=f"/{sugg.name}",
start_position=start_position,
display=display_styled,
display_meta=meta_parts if meta_parts else None,
)
@staticmethod
def _current_slash_token(text: str) -> tuple[str | None, int]:
"""Return ``(token, start_index)`` for the slash token under the cursor.
``token`` is ``None`` when the cursor is not inside a slash command.
``start_index`` is the offset of the leading ``/`` in ``text``.
"""
if not text:
return None, 0
if text.startswith("/"):
# Start-of-buffer slash: complete only while cursor is on the
# command word (before the first space).
space_idx = text.find(" ")
if space_idx != -1:
return None, 0
return text, 0
# Mid-input slash: whitespace + '/' immediately before the cursor.
for i in range(len(text) - 1, -1, -1):
ch = text[i]
if ch == "/":
if i > 0 and not text[i - 1].isspace():
return None, 0
token = text[i:]
if " " in token:
return None, 0
return token, i
if ch.isspace():
return None, 0
return None, 0
try:
from rich.cells import cell_len
from rich.console import Console, Group
from rich.align import Align
from rich.panel import Panel
from rich.table import Table
from rich.text import Text
from rich.markdown import Markdown
from rich.columns import Columns
except ModuleNotFoundError: # pragma: no cover
class Console: # type: ignore
def print(self, *args, **kwargs):
return None
def cell_len(s): # type: ignore
return len(s)
Group = None # type: ignore
Align = None # type: ignore
Panel = None # type: ignore
Table = None # type: ignore
Text = None # type: ignore
Columns = None # type: ignore
class Markdown: # type: ignore
def __init__(self, text: str):
self.text = text
from pathlib import Path
import asyncio
import sys
import json
import threading
import time
from collections import deque
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from src.permissions.types import PermissionAskReply, PermissionAskRequest
from src.agent import Session
from src.config import get_provider_config
from src.outputStyles import resolve_output_style
from src.providers import get_provider_class
from src.providers.anthropic_provider import AnthropicProvider
from src.providers.base import ChatMessage
from src.providers.minimax_provider import MinimaxProvider
from src.services.api.claude import tool_to_api_schema
from src.tool_system.context import ToolContext
from src.tool_system.defaults import build_default_registry
from src.tool_system.protocol import ToolCall
from src.tool_system.renderers import ToolEvent, summarize_tool_result, summarize_tool_use
from src.query.engine import QueryEngine, QueryEngineConfig
from src.query.query import StreamEvent
from src.types.messages import AssistantMessage, SystemMessage, UserMessage
from src.types.content_blocks import TextBlock, ToolUseBlock, ToolResultBlock
# New command system imports
from src.command_system import (
CommandRegistry,
CommandResult,
create_command_context,
execute_command_async,
execute_command_sync,
get_command_registry,
load_and_register_skills,
register_builtin_commands,
)
from src.cost_tracker import CostTracker
from src.history import HistoryLog
from src.repl.at_file_completer import AtFileCompleter
from src.repl.live_status import LiveStatus
try:
from prompt_toolkit.patch_stdout import patch_stdout as _pt_patch_stdout
except ModuleNotFoundError: # pragma: no cover - prompt_toolkit guarded above
from contextlib import nullcontext as _pt_patch_stdout # type: ignore
def _format_edit_summary_text(adds: int, removes: int) -> str:
"""Format an "Added X lines, removed Y lines" summary.
Mirrors the pluralization in the TS reference component
(``FileEditToolUpdatedMessage.tsx``) — sentence-cased standalone
clauses, lowercase ``removed`` after a comma.
"""
if adds <= 0 and removes <= 0:
return ""
parts: list[str] = []
if adds > 0:
parts.append(f"Added {adds} {'line' if adds == 1 else 'lines'}")
if removes > 0:
verb = "Removed" if adds == 0 else "removed"
parts.append(f"{verb} {removes} {'line' if removes == 1 else 'lines'}")
return ", ".join(parts)
# Tool names whose consecutive calls should be coalesced into a single
# ``TaskListV2`` snapshot in the transcript. See
# ``typescript/src/components/TaskListV2.tsx`` for the reference UI.
_TASK_WIDGET_TOOL_NAMES: set[str] = {
"TaskCreate",
"TaskUpdate",
"TaskList",
"TaskGet",
"TodoWrite",
}
class ClawcodexREPL:
"""Interactive REPL for Claw Codex."""
def __init__(
self,
provider_name: str = "glm",
stream: bool = False,
*,
permission_mode: str = "default",
is_bypass_permissions_mode_available: bool = False,
):
# ``is_interactive`` is set during bootstrap phase 2 by
# ``src.init.run_pre_action`` (called from ``cli.main``) before
# the REPL constructor runs. Previously we set it here too,
# but that was the M7.1 gap closed in plan phase 1 of
# ch02-bootstrap. The REPL can rely on
# ``get_is_interactive()`` already being ``True`` by the time
# this constructor runs.
# Stash the resolved permission state so ``ToolContext`` honors
# ``--dangerously-skip-permissions`` / ``--permission-mode`` flags
# passed at startup. See ``src/cli.py:_resolve_permission_state``.
self._permission_mode = permission_mode
self._is_bypass_permissions_mode_available = bool(
is_bypass_permissions_mode_available
)
self.console = Console()
self.provider_name = provider_name
self.stream = stream
# Load configuration
config = get_provider_config(provider_name)
if not config.get("api_key"):
self.console.print("[red]Error: API key not configured.[/red]")
self.console.print("Run [bold]clawcodex login[/bold] to configure.")
sys.exit(1)
# Initialize provider. The persisted /model choice (#280) wins
# over the provider's configured default.
from src.settings.settings import get_persisted_model
provider_class = get_provider_class(provider_name)
self.provider = provider_class(
api_key=config["api_key"],
base_url=config.get("base_url"),
model=get_persisted_model(provider_name) or config.get("default_model")
)
# Create session
self.session = Session.create(
provider_name,
self.provider.model
)
# Late-binding closure: ``tool_context`` is built below, but the
# Agent tool's prompt builder won't read this until much later,
# so reading ``self.tool_context.mcp_clients`` lazily is safe.
def _get_mcp_servers_for_prompt() -> list[str]:
ctx = getattr(self, "tool_context", None)
if ctx is None:
return []
clients = getattr(ctx, "mcp_clients", None) or {}
return list(clients.keys())
self.tool_registry = build_default_registry(
provider=self.provider,
get_available_mcp_servers=_get_mcp_servers_for_prompt,
)
self._engine_messages: list[Any] = []
# C1: build the permission context through the settings loader so
# rules persisted by "always allow" (and hand-written settings
# files) are live in every session — previously this was a bare
# ruleless ToolPermissionContext. Setup warnings (dangerous rules,
# shadowed rules) are intentionally not surfaced yet — phase C6.
from src.permissions.settings_paths import default_setup_paths
from src.permissions.setup import setup_permissions
_perm_setup = setup_permissions(
cwd=str(Path.cwd()),
mode=self._permission_mode, # type: ignore[arg-type]
is_bypass_available=self._is_bypass_permissions_mode_available,
**default_setup_paths(str(Path.cwd())),
)
self.tool_context = ToolContext(
workspace_root=Path.cwd(),
permission_context=_perm_setup.context,
)
self.tool_context.ask_user = self._ask_user_questions
# Permission handler with status control for proper input handling
self._current_status = None
if self._permission_mode == "bypassPermissions":
# The bypass mode short-circuits the registry's permission check
# before the handler is ever consulted; the allow-reply lambda is
# belt-and-braces for any future direct handler caller. (The old
# claim that the doc-write gate calls the handler directly was
# wrong — it flows through check_permissions → registry.)
from src.permissions.types import PermissionAskReply
self.tool_context.allow_docs = True
self.tool_context.permission_handler = (
lambda _request: PermissionAskReply(behavior="allow")
)
else:
self.tool_context.permission_handler = self._handle_permission_request
# Persistent bottom-toolbar accumulators. Mirrors the TS Ink
# status line that always shows model · provider · cwd · turn /
# token totals.
self._stats_turns: int = 0
self._stats_input_tokens: int = 0
self._stats_output_tokens: int = 0
self._direct_stream_abort: bool = False
# Messages the user typed into LiveStatus while the agent was
# working. The main run() loop drains this before falling back to
# ``prompt_session.prompt()`` so queued prompts are sent back-to-back
# without the user having to retype them — matches the TS Ink
# reference's "type while it's still thinking" affordance.
self._queued_prompts: list[str] = []
self._queued_prompts_lock = threading.Lock()
# True only while the main run() loop is blocked on the idle
# ``❯`` prompt. The background notification watcher reads this to
# decide whether it may wake the prompt (so a finished workflow's
# banner + summary surface without a keystroke); it never wakes a
# permission dialog or an in-progress chat turn.
self._at_prompt = False
self._notif_stop: threading.Event | None = None
# Permission dialogs can be requested from different worker paths
# (e.g. subagents/tools). Serialize interactive prompts so we never
# mount competing prompt_toolkit applications at once.
self._permission_prompt_lock = threading.Lock()
# Session-level cache for permission decisions (tool_name -> allow/deny)
# so identical prompts in loops don't repeatedly interrupt the user.
self._permission_decision_cache: dict[str, bool] = {}
# The currently mounted ``LiveStatus`` (if any). ``_safe_input``
# pauses it before reading a synchronous answer (e.g. permission
# prompts) so two prompt_toolkit Applications don't fight over
# the TTY and tear the spinner row.
self._active_live_status: LiveStatus | None = None
# Bounded stash of (label, full_content) pairs for blocks rendered
# truncated in the transcript (currently only Write previews).
# ``ctrl+o`` re-prints the most recent entry as a fresh block
# below — see ``_do_expand_last``. Bounded so the deque doesn't
# grow unboundedly during a long session.
self._expandable_blocks: deque[tuple[str, str]] = deque(maxlen=20)
# Original built-in commands - define this FIRST!
self._original_built_ins = [
"/",
"/help",
"/exit",
"/quit",
"/q",
"/clear",
"/save",
"/load",
"/stream",
"/render-last",
"/tools",
"/tool",
"/skills",
"/init",
"/tui",
]
self._built_in_commands = list(self._original_built_ins)
# Initialize new command system
self._init_command_system()
# Prompt toolkit with tab completion
history_file = Path.home() / ".clawcodex" / "history"
history_file.parent.mkdir(parents=True, exist_ok=True)
# ``_SlashOnlyCompleter`` handles ``/`` slash commands; the
# ``AtFileCompleter`` adds ``@``-mention file completion that
# mirrors the TS Ink reference (see
# ``typescript/src/hooks/fileSuggestions.ts``). Merging keeps
# both behaviors active simultaneously without either side
# interfering with the other's trigger.
from prompt_toolkit.completion import merge_completers
# TTL cache for the slash-command suggestion list. ``build_command
# _suggestions`` walks the user/project/managed skills dirs on every
# call (~1.1s cold), and prompt_toolkit asks the completer on every
# keystroke while typing — so without a cache the first ``/`` press
# blocks the input row for over a second. Refreshed lazily; the
# background warm below populates the cache before the user can
# plausibly press ``/``. Invalidated on a 30 s TTL so newly-added
# skills surface within a turn or two.
self._slash_suggestions_cache: list[Any] | None = None
self._slash_suggestions_cache_at: float = 0.0
self._slash_completer = _SlashOnlyCompleter(
self._get_slash_command_words,
suggestions_provider=self._get_slash_command_suggestions,
)
self._at_completer = AtFileCompleter(
cwd=str(self.tool_context.workspace_root)
)
self.completer = merge_completers(
[self._slash_completer, self._at_completer]
)
# Warm the slash-command suggestion cache in the background so the
# very first ``/`` keystroke doesn't pay the cold import + disk-walk
# cost. Daemon thread so it can't block REPL shutdown.
threading.Thread(
target=self._warm_slash_suggestions_cache,
name="slash-suggestions-warm",
daemon=True,
).start()
# Key bindings.
#
# Multiline-entry contract (mirrors
# ``typescript/src/hooks/useTextInput.ts#handleEnter``):
#
# * plain Enter -> submit
# * Shift+Enter -> insert newline (terminals with
# Kitty-protocol CSI 13;2u, iTerm2
# or VSCode configured via
# /terminal-setup)
# * Meta/Alt/Option+Enter -> insert newline (universally
# supported: the terminal sends
# "\x1b\r", which prompt_toolkit
# parses as Escape+ControlM)
# * ``\`` + Enter -> insert newline (portable fallback
# that works on ANY terminal — the
# trailing backslash is removed and
# replaced by a real newline)
#
# The buffer is always created in ``multiline=True`` mode so that
# real newlines can live in it; we override the default Enter
# behavior below so Enter still submits (prompt_toolkit's default
# in multiline mode is "insert newline").
self.bindings = KeyBindings()
if hasattr(self.bindings, "add"):
@self.bindings.add("/") # type: ignore[attr-defined]
def _show_slash_completions(event): # type: ignore[no-untyped-def]
# Always insert the literal ``/`` — earlier versions
# short-circuited when the buffer was non-empty and
# silently swallowed the keystroke, so paths like
# ``src/repl/core.py`` were untypable. Only auto-pop
# the slash-command menu when ``/`` is the first
# character of the buffer (mirrors the TS reference's
# ``commandSuggestions`` trigger rule).
buf = event.current_buffer
was_empty = buf.text == ""
buf.insert_text("/")
if was_empty:
buf.start_completion(select_first=False)
def _refresh_slash_menu_after_deletion(event, deleter): # type: ignore[no-untyped-def]
# prompt_toolkit's ``complete_while_typing`` only fires on
# ``insert_text`` (buffer.py:1248-1252) — text deletions
# close the completion popup but never reopen it. That's
# what makes ``/exit`` → backspace to ``/ex`` go silent:
# the popup closes when the menu's selected completion no
# longer matches, and nothing re-triggers it. So we
# explicitly restart completion after the deletion when
# the cursor is still on a slash token.
buf = event.current_buffer
deleter(buf)
if not (buf.completer and buf.complete_while_typing()):
return
token, _ = _SlashOnlyCompleter._current_slash_token(
buf.document.text_before_cursor
)
if token is not None:
buf.start_completion(select_first=False)
@self.bindings.add("backspace") # type: ignore[attr-defined]
def _backspace_refreshes_slash_menu(event): # type: ignore[no-untyped-def]
_refresh_slash_menu_after_deletion(
event, lambda b: b.delete_before_cursor(count=1)
)
@self.bindings.add("delete") # type: ignore[attr-defined]
def _delete_refreshes_slash_menu(event): # type: ignore[no-untyped-def]
_refresh_slash_menu_after_deletion(
event, lambda b: b.delete(count=1)
)
@self.bindings.add("c-m") # type: ignore[attr-defined]
def _enter_submits_or_backslash_newline(event): # type: ignore[no-untyped-def]
"""Enter: submit, or convert trailing ``\\`` into a newline.
Exactly mirrors the TypeScript ``handleEnter`` logic. When a
completion popup is open we accept the current selection and
close the popup (prompt_toolkit's default Enter behavior) so
the slash-command menu still works as expected.
"""
buf = event.current_buffer
if buf.complete_state:
buf.complete_state = None
return
text = buf.text
pos = buf.cursor_position
if pos > 0 and text[pos - 1] == "\\":
buf.delete_before_cursor(count=1)
buf.insert_text("\n")
return
buf.validate_and_handle()
@self.bindings.add("escape", "c-m") # type: ignore[attr-defined]
def _meta_or_shift_enter_inserts_newline(event): # type: ignore[no-untyped-def]
"""Meta+Enter (and Kitty-protocol Shift+Enter): insert ``\\n``."""
event.current_buffer.insert_text("\n")
@self.bindings.add("c-o") # type: ignore[attr-defined]
def _expand_last(event): # type: ignore[no-untyped-def]
"""Ctrl+O: re-print the most recent truncated block in
full as a fresh block below the prompt. ``run_in_terminal``
temporarily exits the prompt loop so the output doesn't
fight the prompt's redraw."""
try:
from prompt_toolkit.application import run_in_terminal
run_in_terminal(self._do_expand_last)
except Exception:
# Fallback: print directly. Prompt may redraw oddly
# but at least the expansion lands in scrollback.
self._do_expand_last()
self.prompt_session = PromptSession(
history=FileHistory(str(history_file)),
auto_suggest=AutoSuggestFromHistory(),
completer=self.completer,
style=Style.from_dict({
# Dim background on the ``❯`` marker so the user
# input row reads as a discrete block above the
# transcript — matches Claude Code's input
# background highlight.
'prompt': 'bold fg:ansiblue bg:#262626',
'bottom-toolbar': 'fg:#888888 bg:default',
# Slash-command completion menu (two-column layout:
# /name on the left, [tag] + description on the right).
# Mirrors the TS reference where unselected rows are dim
# and the highlighted row inverts on a tinted background.
'completion-menu': 'bg:default',
'completion-menu.completion': 'fg:#bfbfbf bg:default',
'completion-menu.completion.current': 'fg:#ffffff bg:#005f87 bold',
'completion-menu.meta.completion': 'fg:#7a7a7a bg:default',
'completion-menu.meta.completion.current': 'fg:#dadada bg:#005f87',
'completion.command': 'bold fg:ansigreen',
'completion.tag': 'italic fg:ansicyan',
'completion.description': 'fg:#9a9a9a',
}),
key_bindings=self.bindings,
complete_while_typing=True,
multiline=True,
prompt_continuation=self._prompt_continuation,
bottom_toolbar=self._bottom_toolbar,
)
def _bottom_toolbar(self):
"""Single-line status footer for the input prompt.
Mirrors the TS Ink reference's persistent status row at the
bottom: provider, model, current working directory, and
accumulated turn / token counts for the session. Kept terse so
it doesn't compete with the input row for attention.
"""
try:
provider = (
getattr(self.provider, "provider_name", None)
or self.provider_name
or "?"
)
model = getattr(self.provider, "model", "") or "?"
cwd_full = str(self.tool_context.cwd or self.tool_context.workspace_root)
cwd = self._shorten_path_text(cwd_full) or cwd_full
# Optional advisor segment — appears between cwd and turns
# when ``/advisor`` is set. Mode label (server/client/inactive)
# reflects what the NEXT request will do given the current
# provider + main model, so a stale config under an
# unsupported provider shows "(inactive)" rather than lying.
from src.utils.advisor import format_advisor_status
advisor_seg = format_advisor_status(self.provider, model)
advisor_part = f" {advisor_seg} ·" if advisor_seg else ""
# Advisor token counts — accumulated on the ToolContext
# by ``src/tool_system/tools/advisor.py`` per consultation.
# Surface them next to the worker's counts so the user can
# see how much of the spend went to the reviewer model.
# Hidden when zero so the toolbar stays compact for users
# who haven't enabled the advisor.
adv_in = int(getattr(self.tool_context, "advisor_input_tokens", 0) or 0)
adv_out = int(getattr(self.tool_context, "advisor_output_tokens", 0) or 0)
advisor_tokens = (
f" (advisor: {adv_in} in / {adv_out} out)"
if (adv_in or adv_out) else ""
)
# USD cost — directional estimate based on the upstream
# model's published per-token price. Proxies (litellm,
# openrouter, bedrock) may charge different rates; the
# displayed number is the upstream-list cost, not the
# exact invoice. Hidden when zero (no API turns yet this
# session).
from src.services.pricing import (
compute_session_cost,
format_cost_usd,
)
try:
from src.settings.settings import get_settings as _gs
_settings = _gs()
_advisor_model = (getattr(_settings, "advisor_model", "") or "").strip()
except Exception:
_advisor_model = ""
worker_cost, advisor_cost, total_cost = compute_session_cost(
worker_model=model,
worker_input_tokens=self._stats_input_tokens,
worker_output_tokens=self._stats_output_tokens,
advisor_model=_advisor_model,
advisor_input_tokens=adv_in,
advisor_output_tokens=adv_out,
)
# Space-separated label (matches TUI's "cost N" pattern;
# avoids the REPL/TUI label-style split critic flagged).
cost_part = (
f" · cost {format_cost_usd(total_cost)}"
if total_cost > 0 else ""
)
return (
f" {provider} · {model} · {cwd} ·{advisor_part} "
f"turns: {self._stats_turns} · "
f"tokens: {self._stats_input_tokens} in / "
f"{self._stats_output_tokens} out"
f"{advisor_tokens}"
f"{cost_part} "
)
except Exception:
# Never let the toolbar break the input prompt.
return ""
def _echo_user_input(self, text: str) -> None:
"""Print a user message to the transcript with a dim background.
Used for queued submissions (typed during agent work via
:class:`LiveStatus`) and any other path that needs to surface a
user-authored message into scrollback. Each line is padded to
the terminal width so the highlight reaches the right edge,
matching the boxed input row Claude Code renders for user
messages.
"""
try:
import shutil
width = shutil.get_terminal_size((100, 24)).columns
except Exception:
width = 80
from rich.text import Text
bg_style = "on grey15"
prefix = "❯ "
for idx, line in enumerate(text.split("\n")):
body = (prefix if idx == 0 else " ") + line
padded = body.ljust(max(width, len(body)))
self.console.print(
Text(padded, style=bg_style),
highlight=False,
soft_wrap=True,
)
def _prompt_continuation(self, width, line_number, is_soft_wrap):
"""Continuation prompt for wrapped / multi-line input.
Logical lines get ``"… "`` so it's obvious we're in an in-progress
multi-line prompt; soft wraps get blank padding so long lines
flow naturally. Width-padded to keep the text column aligned
with the primary ``❯ `` prompt.
"""
if is_soft_wrap:
return " " * width
marker = "… "
if width <= len(marker):
return marker[:width]
return marker.rjust(width)
def _ask_user_questions(self, questions: list[dict]) -> dict[str, str]:
# Stop the Rich status spinner if running, so we can get clean input
if self._current_status is not None:
try:
self._current_status.stop()
except Exception:
pass
answers: dict[str, str] = {}
for q in questions:
if isinstance(q, str):
q = {"question": q}
if not isinstance(q, dict):
continue
question_text = str(q.get("question", "")).strip()
options = q.get("options") or []
multi = bool(q.get("multiSelect", False))
if not question_text or not isinstance(options, list) or len(options) < 2:
continue
self.console.print(f"\n[bold]{question_text}[/bold]")
labels: list[str] = []
for i, opt in enumerate(options, start=1):
if isinstance(opt, str):
opt = {"label": opt, "description": ""}
if not isinstance(opt, dict):
continue
label = str(opt.get("label", "")).strip()
desc = str(opt.get("description", "")).strip()
labels.append(label)
self.console.print(f" {i}. {label} [dim]{desc}[/dim]")
other_idx = len(labels) + 1
self.console.print(f" {other_idx}. Other [dim]Provide custom text[/dim]")
prompt = "Select (comma-separated) > " if multi else "Select > "
raw = self._safe_input(prompt).strip()
if not raw:
choice_str = "1"
else:
choice_str = raw
selected: list[str] = []
parts = [p.strip() for p in choice_str.split(",") if p.strip()]
if not parts:
parts = ["1"]
for part in parts:
try:
idx = int(part)
except ValueError:
idx = -1
if idx == other_idx:
free = input("Other > ").strip()
if free:
selected.append(free)
continue
if 1 <= idx <= len(labels):
selected.append(labels[idx - 1])
if not selected:
selected = [labels[0]]
answers[question_text] = ", ".join(selected) if multi else selected[0]
# Restart spinner after getting answers
if self._current_status is not None:
try:
self._current_status.start()
except Exception:
pass
return answers
def _handle_permission_request(
self, request: "PermissionAskRequest"
) -> "PermissionAskReply":
"""Handle interactive permission requests from tools.
C1: receives a :class:`src.permissions.types.PermissionAskRequest`
and returns a :class:`~src.permissions.types.PermissionAskReply`.
The console menu mirrors the TUI modal's option set: allow once /
allow always (when rule suggestions exist) / deny / deny with
feedback — plus the REPL-only ``allow_docs`` enable shortcut.
"""
from src.permissions.types import PermissionAskReply
tool_name = request.tool_name
message = request.message
suggestions = tuple(getattr(request, "suggestions", ()) or ())
with self._permission_prompt_lock:
cache_key = tool_name.strip().lower()
cached = self._permission_decision_cache.get(cache_key)
if cached is not None:
return PermissionAskReply(
behavior="allow" if cached else "deny"
)
# Stop the Rich status spinner if running, so we can get clean input
if self._current_status is not None:
try:
self._current_status.stop()
except Exception:
pass
self.console.print("")
self.console.print("[bold yellow]⚠ Permission Required[/bold yellow]")
self.console.print(f" {message}")
self.console.print("")
# Determine if this is a setting that can be enabled
can_enable_setting = False
setting_to_enable: str | None = None
msg_lower = message.lower()
if "allow_docs" in msg_lower or "documentation files" in msg_lower:
if not self.tool_context.allow_docs:
can_enable_setting = True
setting_to_enable = "allow_docs"
always_label: str | None = None
if suggestions:
from src.permissions.updates import suggestions_label
always_label = suggestions_label(suggestions)
# Build options as (key, description, action) rows so the
# numeric choices always match what was displayed.
options: list[tuple[str, str, str]] = []
if can_enable_setting:
options.append(
("e", f"Enable {setting_to_enable} and allow", "enable")
)
options.append(("y", "Yes, allow this action", "allow"))
if always_label:
options.append(("a", f"Yes, and {always_label}", "always"))
options.append(("n", "No, deny this action", "deny"))
options.append(
("d", "No, and tell Claude what to do differently", "feedback")