"""Corte de archivos TypeScript/JavaScript en trechos por declaración. Hermano del `swift_chunker` y `python_chunker`, para el proyecto Jhonny (`code/`), que es MCP en TypeScript. Antes estos archivos caían en el cortador genérico por ventana de líneas, que graba `symbols: None` — y eso deja vacía la columna que alimenta el lado lexical (pg_trgm) de la busca híbrida. El resultado: buscar `OAuthResourceServer` no rescata ningún `.ts`, porque el único señal que quedaba era el denso (embedding). Aquí el corte sigue la estructura del archivo: el preámbulo (imports + doc de la primera declaración exportada) vira un trecho, y cada declaración de topo (`export class/interface/type/function/const`, o sus variantes sin `export`) vira otro. Los identificadores exportados entran en `symbols` con la misma faixa de `start_line`/`end_line` que el agente usa para leer sólo esa ventana. Tipos de declaración soportados (patrones reales vistos en code/src/): export function buildContentSecurityPolicy(...): string export interface AuthenticatedPrincipal { ... } export type AuthenticationResult = A | B export const HTTP_SECURITY_HEADERS = Object.freeze({ ... }) export class OAuthResourceServer { ... } function bearerToken(...) // sin export sigue importando type JwtVerifier = (...) => Promise<...> """ import os import re # Palabras que abren una declaración a nivel de tope. `type` entra por los # alias de tipo (`export type X = ...`); `const`/`let`/`var` por las # constantes exportadas (ej: `export const HTTP_SECURITY_HEADERS`). _DECL_KW = "class|interface|type|enum|function|const|let|var" # Modificadores que preceden la palabra clave de la declaración: `export` # primero, y después `default`/`async`/`abstract`/`declare`. _MODIFIERS = "export|default|async|abstract|declare|global" _DECL_RE = re.compile( rf"^(?P[ \t]*)(?:(?:{_MODIFIERS})\s+)*(?P{_DECL_KW})" rf"\s+(?P[A-Za-z_$][A-Za-z0-9_$]*)" ) # Declaración exportada con nombre — alimenta `public_symbols` y el tipo # principal del archivo (`file_facts`). _EXPORT_RE = re.compile( rf"^(?:export\s+)(?:(?:{_MODIFIERS})\s+)*(?P{_DECL_KW})" rf"\s+(?P[A-Za-z_$][A-Za-z0-9_$]*)" ) # Declaraciones que vuelven `main_type` del archivo (tipos, no funciones # ni constantes). Sin `export` se ignora: un helper interno no manda. _TYPE_RE = re.compile( rf"^(?:export\s+)(?:(?:{_MODIFIERS})\s+)*(?Pclass|interface|type|enum)" rf"\s+(?P[A-Za-z_$][A-Za-z0-9_$]*)" ) # Nombres declarados en un trecho — alimenta la columna `symbols` (lado # lexical pg_trgm). Casa cualquier `kw name` del cuerpo, incluidos métodos # y constantes internas de una clase, que también son nombres buscables. _SYMBOL_RE = re.compile( rf"\b(?:{_DECL_KW})\s+(?P[A-Za-z_$][A-Za-z0-9_$]*)" ) # Líneas que no abren ni cierran escopo de verdad, para que la contada de # llaves no se engañe con comentario o literal. TS usa comillas simples, # dobles y backticks (template literals): cubrimos las tres. _LINE_COMMENT_RE = re.compile(r"//.*$") _STRING_RE = re.compile( r'"(?:[^"\\]|\\.)*"|\'(?:[^\'\\]|\\.)*\'|`(?:[^`\\]|\\.)*`' ) # Archivos hasta este tamaño vienen un único trecho: el archivo entero. WHOLE_FILE_MAX_LINES = 120 # Teto rígido de caracteres por trecho (nomic-embed-text: 2048 tokens). MAX_CHARS = 5000 # Tamaño que un trecho intenta alcanzar juntando declaraciones vecinas. TARGET_CHARS = 2000 def _strip_noise(line): return _LINE_COMMENT_RE.sub("", _STRING_RE.sub('""', line)) def _preamble_start(lines, i): """Recua de `i` para incluir el doc-comment y anotaciones de la declaración.""" j = i while j > 0: prev = lines[j - 1].strip() if prev.startswith("//") or prev.startswith("*") or prev.startswith("/*") \ or prev.startswith("@"): j -= 1 else: break return j def _split_long(chunk_lines, start_line, max_chars=MAX_CHARS): """Divide un trecho grande en pedazos, siempre en frontera de línea.""" out, buf, buf_start, size = [], [], start_line, 0 for offset, line in enumerate(chunk_lines): if buf and size + len(line) + 1 > max_chars: out.append((buf, buf_start)) buf, buf_start, size = [], start_line + offset, 0 buf.append(line) size += len(line) + 1 if buf: out.append((buf, buf_start)) return out def _merge_small(chunks): """Junta declaraciones vecinas cortas hasta `TARGET_CHARS`. Sólo funde trechos contiguos (el fin de uno encosta el comienzo de otro), para que la faixa de líneas del trecho fundido siga siendo una ventana única y legible con `Read(offset=…, limit=…)`. """ merged = [] for c in chunks: prev = merged[-1] if merged else None contiguous = prev is not None and prev["end_line"] + 1 == c["start_line"] same_kind = prev is not None and prev["kind"] == c["kind"] == "decl" fits = prev is not None and \ len(prev["content"]) + len(c["content"]) + 1 <= TARGET_CHARS if contiguous and same_kind and fits: prev["content"] += "\n" + c["content"] prev["end_line"] = c["end_line"] prev["symbols"] = " ".join( sorted(set(prev["symbols"].split()) | set(c["symbols"].split())) ) else: merged.append(dict(c)) return merged def _symbols_of(text, file_stem, main_type): """Nombres buscables del trecho. El nombre del archivo entra siempre, igual que el tipo principal: es lo que hace que una consulta por el nombre (`OAuthResourceServer`) case con el archivo derecho — exactamente el caso en que la busca densa erraba. """ names = {file_stem} if main_type: names.add(main_type) names.update(m.group("name") for m in _SYMBOL_RE.finditer(text)) return " ".join(sorted(n for n in names if n)) def file_facts(text, rel_path): """Tipo principal, símbolos públicos y resumen — alimenta `file_index`.""" lines = text.splitlines() main_type = None summary = None public_symbols = [] for i, line in enumerate(lines): m = _EXPORT_RE.match(line) if m and m.group("name"): public_symbols.append(m.group("name")) tm = _TYPE_RE.match(line) if tm and main_type is None: main_type = tm.group("name") # Resumen: primera línea del doc-comment justo encima del tipo. for k in range(_preamble_start(lines, i), i): doc = lines[k].strip() if doc.startswith("//") or doc.startswith("*"): summary = doc.lstrip("/").strip().lstrip("*").strip() break if main_type is None: main_type = os.path.splitext(os.path.basename(rel_path))[0] if summary is None: for line in lines[:40]: s = line.strip() if s.startswith("//"): summary = s.lstrip("/").strip() break return { "main_type": main_type, "summary": summary, "public_symbols": sorted(set(public_symbols)), "n_lines": len(lines), } def _depth_at_line_starts(lines): """Profundidad de llaves al INICIO de cada línea.""" depth = 0 depths = [] for line in lines: clean = _strip_noise(line) depths.append(depth) depth += clean.count("{") - clean.count("}") return depths def _top_decls(lines): """Declaraciones de tope: lista de (inicio, fin) en índices medio-abiertos. Una declaración empieza en una línea a profundidad 0 que casa `_DECL_RE` e incluye su preámbulo (doc-comment/anotaciones); va hasta la próxima declaración de tope o el final del archivo. """ depths = _depth_at_line_starts(lines) starts = [] for i, line in enumerate(lines): if depths[i] == 0 and _DECL_RE.match(line): starts.append(_preamble_start(lines, i)) # Normaliza solapamientos (un preámbulo puede meter la declaración previa). cleaned = [] for s in starts: if not cleaned or s > cleaned[-1]: cleaned.append(s) starts = cleaned out = [] for idx, s in enumerate(starts): e = starts[idx + 1] if idx + 1 < len(starts) else len(lines) out.append((s, e)) return out def _body_open(lines, s, e): """Índice de la línea donde se abre el cuerpo `{` de una declaración de tope (primera ida de 0 a 1 dentro de [s, e)). None si no abre cuerpo.""" depth = 0 for i in range(s, e): clean = _strip_noise(lines[i]) opened = clean.count("{") closed = clean.count("}") if depth == 0 and opened > closed: return i depth += opened - closed return None def _member_starts(lines, body_start, decl_end, depths): """Miembros directos (nivel 1) del cuerpo de una clase/interface/enum.""" starts = [] for i in range(body_start + 1, decl_end): if depths[i] != 1: continue if not _DECL_RE.match(lines[i]): continue s = _preamble_start(lines, i) if s > body_start and (not starts or s > starts[-1]): starts.append(s) return starts def chunk_ts(text, rel_path): """Divide un archivo TS/JS en trechos con faixa de líneas y símbolos. Devuelve lista de dicts: content, start_line, end_line, symbols, kind. Las líneas son 1-indexadas e inclusivas en las dos puntas. """ lines = text.splitlines() if not lines: return [] stem = os.path.splitext(os.path.basename(rel_path))[0] main_type = file_facts(text, rel_path)["main_type"] def build(start, end, kind): out = [] for piece, piece_start in _split_long(lines[start:end], start + 1): body = "\n".join(piece).strip() if not body: continue out.append({ "content": body, "start_line": piece_start, "end_line": piece_start + len(piece) - 1, "symbols": _symbols_of(body, stem, main_type), "kind": kind, }) return out if len(lines) <= WHOLE_FILE_MAX_LINES: return build(0, len(lines), "file") top = _top_decls(lines) if not top: return build(0, len(lines), "window") depths = _depth_at_line_starts(lines) chunks = build(0, top[0][0], "header") if top[0][0] > 0 else [] for s, e in top: size = sum(len(lines[i]) + 1 for i in range(s, e)) if size <= TARGET_CHARS: chunks.extend(build(s, e, "decl")) continue # Tipo grande (clase/interface con muchos miembros): corta el cuerpo # por miembro (nivel 1), manteniendo la firma del tipo en el primero. body_open = _body_open(lines, s, e) if body_open is None: chunks.extend(build(s, e, "decl")) continue members = _member_starts(lines, body_open, e, depths) if not members: chunks.extend(build(s, e, "decl")) continue chunks.extend(build(s, members[0], "decl")) for idx, si in enumerate(members): ei = members[idx + 1] if idx + 1 < len(members) else e chunks.extend(build(si, ei, "decl")) return _merge_small(chunks)