cecli/cecli/coders/base_coder.py at main · cecli-dev/cecli · GitHub

Repository navigation

History

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

#!/usr/bin/env python

import asyncio

import base64

import copy

import hashlib

import json

import locale

import math

import mimetypes

import os

import platform

import re

import sys

import threading

import time

import traceback

import weakref

from collections import defaultdict

from datetime import date, datetime

import xxhash

# Optional dependency: used to convert locale codes (eg ``en_US``)

# into human-readable language names (eg ``English``).

try:

from babel import Locale # type: ignore

except ImportError: # Babel not installed – we will fall back to a small mapping

Locale = None

from json.decoder import JSONDecodeError

from pathlib import Path

from typing import List

from urllib.parse import urlparse

from uuid import uuid4 as generate_unique_id

import cecli.prompts.utils.system as prompts

from cecli import __version__, models, urls, utils

from cecli.commands import Commands, SwitchCoderSignal

from cecli.decoding import safe_open

from cecli.exceptions import LiteLLMExceptions

from cecli.helpers import command_parser, command_queue, coroutines, nested, responses

from cecli.helpers.conversation import ConversationService, MessageTag

from cecli.helpers.file_system import FileSystemService

from cecli.helpers.io_proxy import IOProxy

from cecli.helpers.loop_detect import LoopDetectedError, LoopDetector

from cecli.helpers.memory_control import trim_memory

from cecli.helpers.observations.service import ObservationService

from cecli.helpers.profiler import TokenProfiler

from cecli.helpers.sessions import SessionManager

from cecli.helpers.threading import ThreadSafeEvent

from cecli.history import ChatSummary

from cecli.hooks import HookIntegration

from cecli.io import ConfirmGroup, InputOutput

from cecli.linter import Linter

from cecli.llm import litellm

from cecli.mcp import LocalServer

from cecli.reasoning_tags import (

REASONING_TAG,

format_reasoning_content,

remove_reasoning_content,

replace_reasoning_tags,

)

from cecli.repo import ANY_GIT_ERROR, GitRepoProxy

from cecli.repomap import RepoMap

from cecli.report import update_error_prefix

from cecli.run_cmd import run_cmd_async

from cecli.tools.utils.output import print_tool_response

from cecli.tools.utils.registry import ToolRegistry

from cecli.utils import copy_tool_call, format_tokens, is_image_file, safe_relpath

from ..dump import dump # noqa: F401

from ..prompts.utils.registry import PromptObject, PromptRegistry

GLOBAL_DATE = date.today().isoformat()

# Default per-minute token budget used for rate limiting when not configured.

DEFAULT_TOKENS_PER_MINUTE = 1000000

class UnknownEditFormat(ValueError):

def __init__(self, edit_format, valid_formats):

self.edit_format = edit_format

self.valid_formats = valid_formats

super().__init__(

f"Unknown edit format {edit_format}. Valid formats are: {', '.join(valid_formats)}"

)

class MissingAPIKeyError(ValueError):

pass

class FinishReasonLength(Exception):

pass

class EmptyResponseError(Exception):

pass

def wrap_fence(name):

return f"<{name}>", f"</{name}>"

all_fences = [

("`" * 3, "`" * 3),

("`" * 4, "`" * 4), # LLMs ignore and revert to triple-backtick, causing #2879

wrap_fence("source"),

wrap_fence("code"),

wrap_fence("pre"),

wrap_fence("codeblock"),

wrap_fence("sourcecode"),

]

class UsageMeta(type):

"""Metaclass that provides shared accumulator properties across all Coder subclasses.

Every instance shares the same unified total token and cost amounts."""

_total_cost = 0

_total_tokens_sent = 0

_total_tokens_received = 0

_total_cached_tokens = 0

_token_usage_buffer = {}

_token_usage_window = 60.0

@property

def total_cost(cls):

return UsageMeta._total_cost

@total_cost.setter

def total_cost(cls, value):

UsageMeta._total_cost = value

@property

def total_tokens_sent(cls):

return UsageMeta._total_tokens_sent

@total_tokens_sent.setter

def total_tokens_sent(cls, value):

UsageMeta._total_tokens_sent = value

@property

def total_tokens_received(cls):

return UsageMeta._total_tokens_received

@total_tokens_received.setter

def total_tokens_received(cls, value):

UsageMeta._total_tokens_received = value

@property

def total_cached_tokens(cls):

return UsageMeta._total_cached_tokens

@total_cached_tokens.setter

def total_cached_tokens(cls, value):

UsageMeta._total_cached_tokens = value

@classmethod

def _purge_token_usage(cls, model, now=None):

"""Drop a model's token-usage entries older than the rolling window."""

if now is None:

now = time.time()

cutoff = now - UsageMeta._token_usage_window

entries = UsageMeta._token_usage_buffer.get(model, [])

UsageMeta._token_usage_buffer[model] = [

(tokens, ts) for tokens, ts in entries if ts >= cutoff

]

@classmethod

def _record_token_usage(cls, model, delta, now=None):

"""Record a model's request prompt-token usage as (tokens, timestamp)."""

if delta <= 0 or model is None:

return

if now is None:

now = time.time()

UsageMeta._token_usage_buffer.setdefault(model, []).append((delta, now))

UsageMeta._purge_token_usage(model, now)

@classmethod

def _get_token_usage_stats(cls, model, now=None):

"""Return a model's (tokens_last_minute, max_single_request, requests_per_minute)."""

if now is None:

now = time.time()

UsageMeta._purge_token_usage(model, now)

buffer = UsageMeta._token_usage_buffer.get(model, [])

tokens_last_minute = sum(tokens for tokens, _ in buffer)

max_single_request = max((tokens for tokens, _ in buffer), default=0)

requests_per_minute = len(buffer)

return tokens_last_minute, max_single_request, requests_per_minute

@classmethod

def _reset_token_usage(cls, model=None):

"""Clear the rolling token-usage buffer for one model, or all when model is None."""

if model is None:

UsageMeta._token_usage_buffer = {}

else:

UsageMeta._token_usage_buffer.pop(model, None)

class Coder(metaclass=UsageMeta):

# Instance-level properties that delegate to the shared metaclass storage

@property

def total_cost(self):

return type(self).total_cost

@total_cost.setter

def total_cost(self, value):

type(self).total_cost = value

@property

def total_tokens_sent(self):

return type(self).total_tokens_sent

@total_tokens_sent.setter

def total_tokens_sent(self, value):

type(self).total_tokens_sent = value

@property

def total_tokens_received(self):

return type(self).total_tokens_received

@total_tokens_received.setter

def total_tokens_received(self, value):

type(self).total_tokens_received = value

@property

def total_cached_tokens(self):

return type(self).total_cached_tokens

@total_cached_tokens.setter

def total_cached_tokens(self, value):

type(self).total_cached_tokens = value

def _reset_token_usage(self):

"""Clear rolling token usage after restoring a saved session."""

UsageMeta._reset_token_usage()

abs_fnames = None

abs_read_only_fnames = None

abs_read_only_stubs_fnames = None

abs_rules_fnames = None

repo = None

root = "."

primary_root = None

last_coder_commit_hash = None

coder_edited_files = None

last_asked_for_commit_time = 0

repo_map = None

functions = None

num_exhausted_context_windows = 0

num_malformed_responses = 0

last_keyboard_interrupt = None

num_reflections = 0

max_reflections = 3

num_tool_calls = 0

max_tool_calls = 25

turn_count = 0

edit_format = None

file_diffs = True

hashlines = False

yield_stream = False

temperature = None

auto_lint = True

auto_test = False

auto_memory = True

_last_memory_invoke_time = 0.0

test_cmd = None

lint_outcome = None

test_outcome = None

multi_response_content = ""

partial_response_content = ""

partial_response_reasoning_content = ""

partial_response_chunks = []

partial_response_tool_calls = []

partial_response_consolidated = None

commit_before_message = []

message_cost = 0.0

message_tokens_sent = 0

message_tokens_received = 0

message_cached_tokens = 0

message_cost_deferred = None

add_cache_headers = False

cache_warming_thread = None

num_cache_warming_pings = 0

suggest_shell_commands = True

detect_urls = True

ignore_mentions = None

chat_language = None

commit_language = None

file_watcher = None

mcp_manager = None

compact_context_completed = True

_drain_queue_after_command = False

suppress_announcements_for_next_prompt = False

tool_reflection = False

last_user_message = ""

uuid: str = ""

parent_uuid: str = ""

model_kwargs = {}

cost_multiplier = 1

stop_on_empty = True

error_code = None

_output_loop_detected = False

_output_loop_message = ""

# Task coordination state variables

input_running = False

output_running = False

# Context management settings (for all modes)

context_management_enabled = False # Disabled by default except for agent mode

large_file_token_threshold = (

25000 # Files larger than this will be truncated when context management is enabled

)

ok_to_warm_cache = False

# Weak reference to TUI app instance (when running in TUI mode)

tui = None

_prompt_cache = {}

@classmethod

async def create(

self,

main_model=None,

edit_format=None,

io=None,

from_coder=None,

summarize_from_coder=True,

args=None,

**kwargs,

import cecli.coders as coders

if not main_model:

if from_coder:

main_model = from_coder.main_model

else:

main_model = models.Model(models.DEFAULT_MODEL_NAME, io=io)

if edit_format == "code":

edit_format = main_model.edit_format

elif edit_format is None:

if from_coder:

edit_format = from_coder.edit_format

else:

edit_format = main_model.edit_format

if not io and from_coder:

io = from_coder.io

if from_coder:

if not args:

args = from_coder.args

use_kwargs = dict(from_coder.original_kwargs) # copy orig kwargs

update = dict(

fnames=list(from_coder.abs_fnames),

read_only_fnames=list(from_coder.abs_read_only_fnames), # Copy read-only files

read_only_stubs_fnames=list(

from_coder.abs_read_only_stubs_fnames

), # Copy read-only stubs

rules_fnames=list(from_coder.abs_rules_fnames), # Copy rules files

done_messages=[],

cur_messages=[],

coder_commit_hashes=from_coder.coder_commit_hashes,

commands=from_coder.commands.clone(),

ignore_mentions=from_coder.ignore_mentions,

file_watcher=from_coder.file_watcher,

mcp_manager=from_coder.mcp_manager,

registered_tools=copy.deepcopy(from_coder.registered_tools),

registered_servers=copy.deepcopy(from_coder.registered_servers),

auto_memory=from_coder.auto_memory,

uuid=from_coder.uuid,

parent_uuid=from_coder.parent_uuid,

repo=from_coder.repo,

primary_root=from_coder.primary_root,

summarizer=from_coder.summarizer,

)

use_kwargs.update(update) # override to complete the switch

use_kwargs.update(kwargs) # override passed kwargs

kwargs = use_kwargs

from_coder.ok_to_warm_cache = False

res = None

if (

getattr(main_model, "copy_paste_mode", False)

and getattr(main_model, "copy_paste_transport", "api") == "clipboard"

res = coders.CopyPasteCoder(main_model, io, args=args, **kwargs)

if not res:

coder_name = coders.EDIT_FORMAT_MAP.get(edit_format)

if coder_name:

coder_cls = getattr(coders, coder_name)

res = coder_cls(main_model, io, args=args, **kwargs)

if res is not None:

if from_coder:

# Preserve TUI ref in all child coders

if from_coder.tui:

res.tui = from_coder.tui

# The scheduler is an active service owned by the primary

# coder: a switch runs inside the same event loop, so carry it

# over (and rebind its coder) instead of leaking a ticking

# zombie onto the retired coder. Recreations that keep the

# uuid are switches/reloads; sub-agents spawn with a fresh

# uuid, and a stopped scheduler (reload) is never handed on.

scheduler = getattr(from_coder, "scheduler", None)

if scheduler is not None and scheduler.running and res.uuid == from_coder.uuid:

from_coder.scheduler = None

res.scheduler = scheduler

scheduler.coder = res

# Sub-agents get a dedicated, independent MCP manager so they

# can rebuild a custom tool list (their own LocalServer tools /

# filters) and be disconnected independently from the parent.

# The parent's server configs are copied into a fresh manager

# whose connections are created on this loop; the Local server

# is left to initialize_mcp_tools() so it is recreated from the

# sub-agent's filters.

if (

from_coder.mcp_manager

and res.uuid

and res.parent_uuid

and res.parent_uuid != res.uuid

res.mcp_manager = await from_coder.mcp_manager.spawn_child(

io=IOProxy.unwrap(res.io)

)

if res.mcp_manager:

# When switching to a non-agent coder, disconnect the "Local" MCP server

# (which provides agent-only tools like tool calling and file editing)

# so it's not available in non-agent modes.

if not isinstance(res, coders.AgentCoder):

local_server = res.mcp_manager.get_server("Local")

if local_server and local_server.is_connected:

await res.mcp_manager.disconnect_server("Local")

if res.uuid == from_coder.uuid:

res.prompt_queue = from_coder.prompt_queue.copy()

res._queue_counter = from_coder._queue_counter

await res.initialize_mcp_tools()

# Store only small/primitive kwargs to avoid retaining large object references.

# Large objects (repo, mcp_manager, commands, summarizer, file_watcher, etc.)

# are either overridden during clone() or accessible from instance attributes.

_LARGE_KWARGS = {"repo", "mcp_manager", "commands", "summarizer", "file_watcher"}

res.original_kwargs = {k: v for k, v in kwargs.items() if k not in _LARGE_KWARGS}

return res

valid_formats = list(coders.EDIT_FORMAT_MAP.keys())

raise UnknownEditFormat(edit_format, valid_formats)

async def clone(self, **kwargs):

new_coder = await Coder.create(from_coder=self, **kwargs)

return new_coder

def __init__(

self,

main_model,

io,

args=None,

repo=None,

fnames=None,

add_gitignore_files=False,

read_only_fnames=None,

read_only_stubs_fnames=None,

rules_fnames=None,

show_diffs=False,

auto_commits=True,

dirty_commits=True,

dry_run=False,

map_tokens=1024,

verbose=False,

stream=True,

use_git=True,

cur_messages=None,

done_messages=None,

auto_lint=True,

auto_test=False,

auto_memory=True,

lint_cmds=None,

test_cmd=None,

coder_commit_hashes=None,

map_mul_no_files=8,

map_max_line_length=100,

commands=None,

summarizer=None,

map_refresh="auto",

cache_prompts=False,

num_cache_warming_pings=0,

suggest_shell_commands=True,

chat_language=None,

commit_language=None,

detect_urls=True,

ignore_mentions=None,

file_watcher=None,

auto_copy_context=False,

auto_accept_architect=True,

mcp_manager=None,

enable_context_compaction=False,

max_compaction_retries=3,

context_compaction_max_tokens=None,

context_compaction_summary_tokens=8192,

map_cache_dir=".",

repomap_in_memory=False,

linear_output=False,

security_config=None,

registered_tools=None,

registered_servers=None,

uuid: str = "",

parent_uuid: str = "",

root=None,

primary_root=None,

init_metadata={},

from cecli.helpers.agents.service import AgentService

# initialize from args.map_cache_dir

self.coroutines = coroutines

# Per-instance tool and server filtering dictionaries

# Each contains "included" and "excluded" sets that filter from the global singletons

self.registered_tools = {"included": set(), "excluded": set()}

self.registered_servers = {"included": set(), "excluded": set()}

self._inherited_tools = False

if registered_tools is not None or registered_servers is not None:

self.registered_tools = registered_tools

self.registered_servers = registered_servers

self._inherited_tools = True

self.interrupt_event = ThreadSafeEvent()

self.uuid = str(generate_unique_id()).split("-")[0]

self.reflected_message = None

if uuid:

self.uuid = str(uuid).split("-")[0]

if parent_uuid:

self.parent_uuid = str(parent_uuid).split("-")[0]

self.map_cache_dir = map_cache_dir

self.chat_language = chat_language

self.commit_language = commit_language

self.commit_before_message = []

self.coder_commit_hashes = set()

self.rejected_urls = set()

self.abs_root_path_cache = {}

self.auto_copy_context = auto_copy_context

self.auto_accept_architect = auto_accept_architect

try:

self.security_config = json.loads(security_config)

except (json.JSONDecodeError, TypeError):

self.security_config = {}

self.ignore_mentions = ignore_mentions

if not self.ignore_mentions:

self.ignore_mentions = set()

self.file_watcher = file_watcher

if self.file_watcher:

self.file_watcher.coder = self

self.suggest_shell_commands = suggest_shell_commands

self.detect_urls = detect_urls

self.args = args

# Init metadata should not persist between initializations

self.init_metadata = {}

self.num_cache_warming_pings = num_cache_warming_pings

self.mcp_manager = mcp_manager

self.enable_context_compaction = enable_context_compaction

self.context_compaction_current_ratio = 0

self.context_compaction_max_tokens = context_compaction_max_tokens

self.context_compaction_summary_tokens = context_compaction_summary_tokens

self.max_compaction_retries = max_compaction_retries

self.max_reflections = nested.getter(self.args, "max_reflections", 3)

self.max_tool_calls = nested.getter(self.args, "max_tool_calls", 25)

if not fnames:

fnames = []

if io is None:

io = InputOutput()

if coder_commit_hashes:

self.coder_commit_hashes = coder_commit_hashes

else:

self.coder_commit_hashes = set()

self.chat_completion_call_hashes = []

self.chat_completion_response_hashes = []

self.need_commit_before_edits = set()

self.message_tokens_sent = 0

self.message_tokens_received = 0

self.message_cached_tokens = 0

self.token_profiler = TokenProfiler(

enable_printing=nested.getter(self.args, "show_speed", False)

)

self.verbose = verbose

self.abs_fnames = set()

self.abs_read_only_fnames = set()

self.add_gitignore_files = add_gitignore_files

self.abs_read_only_stubs_fnames = set()

self.abs_rules_fnames = set()

self.io = io

# Wrap io with IOProxy for coder_uuid injection in output messages

# Always create a new IOProxy so sub-agents get their own _coder_uuid.

# Unwrap any existing IOProxy to avoid fragile nested proxy chains.

raw_io = IOProxy.unwrap(io)

self.io = IOProxy(raw_io, self)

if not self.parent_uuid:

self.io.coder = weakref.ref(self)

self.manual_copy_paste = (

nested.getter(main_model, "copy_paste_transport", "api") == "clipboard"

)

self.copy_paste_mode = (

nested.getter(main_model, "copy_paste_mode", False) or auto_copy_context

)

self.shell_commands = []

self.partial_response_tool_calls = []

if not auto_commits:

dirty_commits = False

self.auto_commits = auto_commits

self.dirty_commits = dirty_commits

self.dry_run = dry_run

self.pretty = self.io.pretty

self.linear_output = linear_output

self.io.linear = linear_output

self.main_model = main_model

# Set the reasoning tag name based on model settings or default

self.reasoning_tag_name = (

self.get_active_model().reasoning_tag

if self.get_active_model().reasoning_tag

else REASONING_TAG

)

self.stream = stream and self.get_active_model().streaming and not self.manual_copy_paste

if cache_prompts and self.get_active_model().cache_control:

self.add_cache_headers = True

self.show_diffs = show_diffs

# Initialize all registry sub systems

AgentService.get_instance(self)

ConversationService.get_chunks(self).initialize_conversation_system()

ObservationService.get_instance(self)

self.commands = commands or Commands(self.io, self, args=args)

self.commands.coder = self

# Prompt queue for CLI-33: in-memory FIFO queue for deferred prompt

# processing. The queue lives on the coder so primary agents and

# sub-agents each have their own independent queue, managed by

# cecli.helpers.command_queue.

self.prompt_queue = []

self._queue_counter = 0

self._queue_lock = threading.Lock()

self._processing_queue = False

self._active_queue_prompt = None

self._drain_queue_after_command = False

self.data_cache = {

"repo": {"last_key": "", "read_only_count": None},

}

self.repo = repo

if use_git and self.repo is None:

try:

self.repo = GitRepoProxy.for_root(

None,

self.io,

fnames=fnames,

git_dname=None,

models=main_model.commit_message_models(),

)

except FileNotFoundError:

pass

if self.repo:

self.root = self.repo.root

for fname in fnames:

fname = Path(fname)

if self.repo and self.repo.git_ignored_file(fname) and not self.add_gitignore_files:

self.io.tool_warning(f"Skipping {fname} that matches gitignore spec.")

continue

if self.repo and self.repo.ignored_file(fname) and not self.add_gitignore_files:

self.io.tool_warning(f"Skipping {fname} that matches cecli.ignore spec.")

continue

if not fname.exists():

if utils.touch_file(fname):

self.io.tool_output(f"Creating empty file {fname}")

else:

self.io.tool_warning(f"Can not create {fname}, skipping.")

continue

if not fname.is_file():

self.io.tool_warning(f"Skipping {fname} that is not a normal file.")

continue

fname = str(fname.resolve())

self.abs_fnames.add(fname)

self.check_added_files()

if not self.repo:

self.root = utils.find_common_root(self.abs_fnames)

# Allow sub-agent classes to override the working root (multi-project workspaces).

if root is not None:

self.root = os.path.normpath(os.path.abspath(root))

# A sub-agent may operate on a different base path than its parent. In

# that case its repo must be scoped to *this* root (per-base-path),

# rather than inheriting the parent's repo, so the coder's repo/fs match

# its own root.

if use_git and self.repo is not None and os.path.normpath(self.repo.root) != self.root:

try:

self.repo = GitRepoProxy.for_root(

self.root,

self.io,

fnames=[self.root],

git_dname=None,

models=main_model.commit_message_models(),

)

except FileNotFoundError:

self.repo = None

# Store the root of the primary coder so skills files, custom tools, and

# sub-agent paths can be resolved relative to the primary workspace even

# when this coder operates on a different base path.

self.primary_root = (

primary_root

if primary_root is not None

else getattr(self, "primary_root", None) or self.root

)

# Initialize the per-base-path FileSystemService for this coder

self.fs = FileSystemService.for_root(

root=self.root if hasattr(self, "root") else ".",

repo=self.repo if hasattr(self, "repo") else None,

)

# Auto-return the per-root service (and its git repo) when the last

# coder sharing this base path is destroyed.

_fs_key = FileSystemService._normalize_root(self.root if hasattr(self, "root") else ".")

FileSystemService._inc_ref(_fs_key)

weakref.finalize(self, FileSystemService._release, _fs_key)

if read_only_fnames:

self.abs_read_only_fnames = set()

for fname in read_only_fnames:

abs_fname = self.abs_root_path(fname)

if os.path.exists(abs_fname):

self.abs_read_only_fnames.add(abs_fname)

else:

if verbose:

self.io.tool_warning(

f"Error: Read-only file {fname} does not exist. Skipping."

)

if read_only_stubs_fnames:

self.abs_read_only_stubs_fnames = set()

for fname in read_only_stubs_fnames:

abs_fname = self.abs_root_path(fname)

if os.path.exists(abs_fname):

self.abs_read_only_stubs_fnames.add(abs_fname)

else:

if verbose:

self.io.tool_warning(

f"Error: Read-only (stub) file {fname} does not exist. Skipping."

)

if rules_fnames:

self.abs_rules_fnames = set()

for fname in rules_fnames:

abs_fname = self.abs_root_path(fname)

if os.path.exists(abs_fname):

self.abs_rules_fnames.add(abs_fname)

else:

if verbose:

self.io.tool_warning(f"Error: Rules file {fname} does not exist. Skipping.")

if map_tokens is None:

use_repo_map = main_model.use_repo_map

map_tokens = 1024

else:

use_repo_map = map_tokens > 0

max_inp_tokens = self.get_active_model().info.get("max_input_tokens") or 0

has_map_prompt = nested.getter(self, "gpt_prompts.repo_content_prefix")

if use_repo_map and self.repo and has_map_prompt:

repo_root = self.root

self.repo_map = RepoMap(

map_tokens,

self.map_cache_dir,

self.get_active_model(),

io,

self.gpt_prompts.repo_content_prefix,

self.verbose,

max_inp_tokens,

map_mul_no_files=map_mul_no_files,

refresh=map_refresh,

max_code_line_length=map_max_line_length,

repo_root=repo_root,

use_memory_cache=repomap_in_memory,

use_enhanced_map=getattr(self.args, "use_enhanced_map", False),

)

self.summarizer = summarizer or ChatSummary(

[self.get_active_model().weak_model, self.get_active_model()],

self.get_active_model().max_chat_history_tokens,

)

self.summarizer_thread = None

self.summarized_done_messages = []

self.summarizing_messages = None

self.files_edited_by_tools = set()

# Linting and testing

self.linter = Linter(

root=self.root, encoding=io.encoding, interrupt_event=self.interrupt_event

)

self.auto_lint = auto_lint

self.setup_lint_cmds(lint_cmds)

self.lint_cmds = lint_cmds

self.auto_test = auto_test

self.auto_memory = auto_memory

self.test_cmd = test_cmd

# Clean up todo list file on startup; sessions will restore it when needed

todo_file_path = self.local_agent_folder("todo.txt")

abs_path = self.abs_root_path(todo_file_path)

if os.path.isfile(abs_path):

try:

os.remove(abs_path)

if self.verbose:

self.io.tool_output(f"Removed existing todo list file: {todo_file_path}")

except Exception as e:

self.io.tool_warning(f"Could not remove todo list file {todo_file_path}: {e}")

customizations = dict()

try:

if self.args:

customizations = nested.getter(self.args, "custom", "{}")

customizations = json.loads(customizations)

except (json.JSONDecodeError, TypeError):

customizations = dict()

pass

self.custom = customizations

self.file_diffs = nested.getter(self.args, "file_diffs", True)

if nested.getter(self.custom, "prompt_map.all", None):

prompts = PromptRegistry.get_prompt(nested.getter(self.custom, "prompt_map.all"))

prompt_obj = PromptObject(prompts)

Coder._prompt_cache[self.prompt_format] = prompt_obj

if nested.getter(self.custom, f"prompt_map.{self.prompt_format}", None):

prompts = PromptRegistry.get_prompt(

nested.getter(self.custom, f"prompt_map.{self.prompt_format}")

)

prompt_obj = PromptObject(prompts)

Coder._prompt_cache[self.prompt_format] = prompt_obj

# validate the functions jsonschema

if self.functions:

from jsonschema import Draft7Validator

for function in self.functions:

Draft7Validator.check_schema(function)

if self.verbose:

self.io.tool_output("JSON Schema:")

self.io.tool_output(json.dumps(self.functions, indent=4))

self.post_init()

def post_init(self):

pass

@property

def gpt_prompts(self):

"""Get prompts from the registry based on the coder type."""

cls = self.__class__

# Every coder class MUST have a prompt_format attribute

if not hasattr(cls, "prompt_format"):

raise AttributeError(

f"Coder class {cls.__name__} must have a 'prompt_format' attribute. "

"Add 'prompt_format = \"<format_name>\"' to the class definition."

)

if cls.prompt_format is None:

raise AttributeError(

f"Coder class {cls.__name__} has prompt_format=None. "

"It must have a valid prompt format name."

)

prompt_name = cls.prompt_format

# Check cache first

if prompt_name in Coder._prompt_cache:

return Coder._prompt_cache[prompt_name]

# Get prompts from registry

prompts = PromptRegistry.get_prompt(prompt_name)

# Cache the prompt object

prompt_obj = PromptObject(prompts)

Coder._prompt_cache[prompt_name] = prompt_obj

return prompt_obj

@property

def done_messages(self):

"""Get DONE messages from ConversationManager."""

return ConversationService.get_manager(self).get_messages_dict(MessageTag.DONE)

@property

def cur_messages(self):

"""Get CUR messages from ConversationManager."""

return ConversationService.get_manager(self).get_messages_dict(MessageTag.CUR)

@staticmethod

def _strip_provider(model_name: str) -> str:

"""Remove provider prefix from model name (e.g., 'openai/gpt-4' -> 'gpt-4')."""

if "/" in model_name:

return model_name.split("/", 1)[1]

return model_name

def get_announcements(self):

sections = {}

# --- MODELS ---

main_model = self.main_model

models_items = [f"{self._strip_provider(main_model.name)} (main)"]

agent_model = main_model.agent_model

weak_model = main_model.weak_model

if agent_model and agent_model.name != main_model.name:

models_items.append(f"{self._strip_provider(agent_model.name)} (agent)")

if weak_model and weak_model.name != main_model.name:

models_items.append(f"{self._strip_provider(weak_model.name)} (weak)")

if self.edit_format == "architect":

models_items.append(f"{self._strip_provider(main_model.editor_model.name)} (editor)")

sections["Models"] = {"items": models_items}

# --- SETTINGS ---

settings_items = []

# Edit format

settings_items.append(f"{self.edit_format} (edit format)")

# Thinking tokens

thinking_tokens = self.get_active_model().get_thinking_tokens()

if thinking_tokens:

settings_items.append(f"{thinking_tokens} think tokens")

# Reasoning effort

reasoning_effort = self.get_active_model().get_reasoning_effort()

if reasoning_effort:

settings_items.append(f"reasoning {reasoning_effort}")

# Prompt cache

View remainder of file in raw view

cecli/cecli/coders/base_coder.py at main · cecli-dev/cecli · GitHub

Navigation Menu

Repository navigation

Expand file tree

Breadcrumbs

base_coder.py

Latest commit

History

Breadcrumbs

base_coder.py

File metadata and controls

Footer

Footer navigation