File size: 9,924 Bytes
2695555
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""5.1: o terminal no nível das CLIs de agente. -p (texto e JSON, stdin, código de saída, sem sobrescrever a conversa
salva), modo plano, "sempre nesta sessão" (nunca para apagar em massa), ferramentas de código só no terminal e nunca
para o Gargantua, edição com diff e /undo, AGENTS.md e /init, /compact, /mode e /resume. Sem Ollama de verdade."""
import contextlib
import io
import json
import os
import sys
import tempfile
from pathlib import Path

pasta = Path(tempfile.mkdtemp(prefix="opentars-cli-"))
os.environ["TARS_ARQUIVO_SESSOES"] = str(pasta / "sessoes.json")
os.environ["HOME"] = str(pasta / "casa")
(pasta / "casa").mkdir()

from _util import carregar_tars  # noqa: E402

tars = carregar_tars()
tars.ollama_online = lambda: False
tars.avisos_de_ambiente = lambda: []

# ---------------------------------------------------------------- -p
sessao_salva = json.dumps({"versao": 2, "conversa": [{"role": "user", "content": "conversa da pessoa"}]})
Path(os.environ["TARS_ARQUIVO_SESSOES"]).write_text(sessao_salva)
recebidos = []


def processar_falso(texto, anexos=None):
    recebidos.append(texto)
    tars.emitir("ferramenta", nome="execute_terminal", resumo="ls")
    tars.emitir("ferramenta_fim", nome="execute_terminal", sucesso=True, segundos=0.1)
    print("passo que não pode sujar o stdout")
    tars.salvar_sessoes()
    return "três arquivos"


tars.processar_mensagem = processar_falso


def rodar_print(pedido, formato="text", entrada="", modelo=None):
    saida, erros = io.StringIO(), io.StringIO()
    stdin_antigo = sys.stdin
    sys.stdin = io.StringIO(entrada)
    try:
        with contextlib.redirect_stdout(saida), contextlib.redirect_stderr(erros):
            codigo = tars.modo_print(pedido, formato, modelo)
    finally:
        sys.stdin = stdin_antigo
    return codigo, saida.getvalue(), erros.getvalue()


codigo, saida, erros = rodar_print("quantos arquivos?", entrada="a.txt\nb.txt\n")
assert codigo == 0 and saida == "três arquivos\n", (codigo, saida)
assert "passo que não pode sujar" in erros and "passo" not in saida
assert "<stdin>\na.txt\nb.txt" in recebidos[-1], recebidos[-1]
assert Path(os.environ["TARS_ARQUIVO_SESSOES"]).read_text() == sessao_salva, "o -p não pode sobrescrever a conversa salva"
codigo, saida, _ = rodar_print("liste", "json", modelo="qwen3:4b")
dados = json.loads(saida)
assert codigo == 0 and dados["ok"] and dados["resposta"] == "três arquivos" and tars.modelo_manual == "qwen3:4b"
assert dados["ferramentas"] == [{"nome": "execute_terminal", "resumo": "ls", "sucesso": True}], dados
tars.processar_mensagem = lambda texto, anexos=None: None
assert rodar_print("algo")[0] == 1, "sem resposta: código 1"
assert rodar_print("")[0] == 2, "sem pedido nem stdin: código 2"
print("OK: -p em texto e JSON, stdin como dado, passos no stderr, código de saída e conversa salva intacta")

# ---------------------------------------------------------------- modo plano
chamou = []
tars._DISPATCH_FERRAMENTAS["open_application"] = lambda a: chamou.append("abriu") or {"sucesso": True, "mensagem": "ok"}
tars._DISPATCH_FERRAMENTAS["list_windows"] = lambda a: chamou.append("olhou") or {"sucesso": True, "mensagem": "janelas"}
tars._tratar_comando_do_chat("/modo plano")
assert tars.MODO_PERMISSAO == "plano"
r = tars.executar_ferramenta("open_application", {"app": "firefox"})
assert r["sucesso"] and "NOT run" in r["mensagem"] and chamou == [] and tars.PASSOS_DO_PLANO == ["open_application firefox"]
assert tars.executar_ferramenta("list_windows", {})["mensagem"] == "janelas" and chamou == ["olhou"], "só olhar continua"
with contextlib.redirect_stdout(io.StringIO()) as plano:
    tars._mostrar_plano()
assert "open_application firefox" in plano.getvalue() and not tars.PASSOS_DO_PLANO
tars._tratar_comando_do_chat("/mode auto")
assert tars.MODO_PERMISSAO == "automatico"
tars._tratar_comando_do_chat("/modo perguntar")
print("OK: no modo plano nada que muda roda (o que só olha roda) e o plano aparece no fim")

# ---------------------------------------------------------------- "sempre nesta sessão"
respostas = []
tars.input = lambda prompt="": respostas.pop(0)
tars.confirmar_comando_hook = None
respostas[:] = ["sempre"]
assert tars._confirmar_comando_perigoso("git push origin main") and not respostas
assert tars._confirmar_comando_perigoso("git push origin main"), "liberado nesta sessão: não pergunta de novo"
respostas[:] = ["sempre", "n"]
assert not tars._confirmar_comando_perigoso("rm -rf ~/pasta"), "'sempre' não vale para apagar em massa"
assert not tars._confirmar_comando_perigoso("rm -rf ~/pasta") and not respostas, "e pergunta de novo"
print("OK: 'sempre nesta sessão' vale para envio/instalação; apagar em massa pergunta toda vez")

# ---------------------------------------------------------------- ferramentas de código
nomes = lambda lista: {f["function"]["name"] for f in lista}  # noqa: E731
tars.CLI_ATIVO = False
assert "edit_file" not in nomes(tars.ferramentas_do_pedido("codigo", "corrige o bug no app.py")), "fora do terminal, não"
tars.CLI_ATIVO = True
assert {"read_file", "search_files", "edit_file", "write_file"} <= nomes(tars.ferramentas_do_pedido(None, "corrige o bug no app.py"))
assert "edit_file" not in nomes(tars.ferramentas_do_pedido(None, "abre a calculadora")), "pedido que não é de código"
assert "edit_file" not in nomes(tars.ferramentas_do_modelo("ProjectOpenTARS/Gargantua:code", "codigo", "corrige o app.py"))
assert len(tars.ferramentas_do_modelo("ProjectOpenTARS/Gargantua:latest", "codigo", "corrige o app.py")) == 19

projeto = pasta / "projeto"
projeto.mkdir()
os.chdir(projeto)
(projeto / "app.py").write_text("def soma(a, b):\n    return a - b\n\nprint(soma(1, 2))\n")
r = tars.executar_ferramenta("read_file", {"path": "app.py"})
assert "    2      return a - b" in r["mensagem"], r["mensagem"]
assert "app.py:2" in tars.executar_ferramenta("search_files", {"pattern": "return a"})["mensagem"]
tars.iniciar_rodada()
respostas[:] = ["s"]
r = tars.executar_ferramenta("edit_file", {"path": "app.py", "old_text": "return a - b", "new_text": "return a + b"})
assert r["sucesso"] and "return a + b" in (projeto / "app.py").read_text() and not respostas, r
respostas[:] = ["n"]
r = tars.executar_ferramenta("write_file", {"path": "novo.py", "content": "x = 1\n"})
assert not r["sucesso"] and not (projeto / "novo.py").exists(), "negado: nada é escrito"
assert not tars.executar_ferramenta("edit_file", {"path": "app.py", "old_text": "a", "new_text": "b"})["sucesso"], \
    "trecho repetido é recusado"
tars.iniciar_rodada()
tars.MODO_PERMISSAO = "automatico"
r = tars.executar_ferramenta("write_file", {"path": "novo.py", "content": "x = 1\n"})
assert r["sucesso"] and (projeto / "novo.py").read_text() == "x = 1\n", "modo automático: dentro do projeto não pergunta"
fora = pasta / "casa" / "fora.txt"
respostas[:] = ["n"]
assert not tars.executar_ferramenta("write_file", {"path": str(fora), "content": "x"})["sucesso"] and not fora.exists(), \
    "fora da pasta do projeto pergunta mesmo no automático"
tars.MODO_PERMISSAO = "perguntar"
assert tars.desfazer_ultima_rodada() == [str(projeto / "novo.py")] and not (projeto / "novo.py").exists()
with contextlib.redirect_stdout(io.StringIO()):
    tars._tratar_comando_do_chat("/undo")
assert "return a - b" in (projeto / "app.py").read_text(), "/undo volta a edição do pedido anterior"
assert tars.desfazer_ultima_rodada() == []
print("OK: ferramentas de código só no terminal e nunca para o Gargantua; diff com confirmação; /undo volta")

# ---------------------------------------------------------------- AGENTS.md e /init
(projeto / "README.md").write_text("# Calculadora\n")
(projeto / "test_app.py").write_text("def test(): pass\n")
with contextlib.redirect_stdout(io.StringIO()):
    tars._tratar_comando_do_chat("/init")
texto = (projeto / "AGENTS.md").read_text()
assert texto.startswith("# Calculadora") and "python3 -m pytest" in texto and "Python" in texto, texto
assert tars.criar_agents_md() is None, "não sobrescreve"
(projeto / "sub").mkdir()
assert tars.carregar_contexto_projeto(projeto / "sub") == projeto / "AGENTS.md", "acha o da pasta de cima"
assert "PROJECT INSTRUCTIONS" in tars._memoria_para_prompt()
tars.CLI_ATIVO = False
assert "PROJECT INSTRUCTIONS" not in tars._memoria_para_prompt(), "só na conversa do terminal"
tars.CLI_ATIVO = True
print("OK: /init cria o AGENTS.md (sem sobrescrever) e ele entra no prompt só no terminal")

# ---------------------------------------------------------------- /compact e /context
tars.conversa[:] = [{"role": "system", "content": "s"}]
for i in range(6):
    tars.conversa += [{"role": "user", "content": f"pedido {i}"}, {"role": "assistant", "content": f"resposta {i}"}]
antes, depois = tars.compactar_conversa(resumir=lambda txt: "resumo: " + txt.count("pedido") * "x")
usuarios = [m["content"] for m in tars.conversa if m["role"] == "user"]
assert usuarios[0].startswith("[Summary of the earlier conversation]\nresumo: xxx") and usuarios[1:] == ["pedido 3", "pedido 4", "pedido 5"]
assert depois < antes and tars.conversa[0]["role"] == "system"
assert tars.compactar_conversa() is None, "conversa curta: nada a compactar"
print("OK: /compact troca o começo por um resumo e guarda os 3 últimos pedidos inteiros")

# ---------------------------------------------------------------- /resume
import tars_historico  # noqa: E402

h = tars_historico.Historico(str(pasta / "historico.json"))
tars.SESSAO_CLI["historico"] = h
h.salvar("antiga", [{"role": "user", "content": "como faço backup"}, {"role": "assistant", "content": "assim"}])
respostas[:] = ["1"]
with contextlib.redirect_stdout(io.StringIO()):
    assert tars._escolher_conversa()
assert tars.SESSAO_CLI["id"] == "antiga" and tars.conversa[1]["content"] == "como faço backup"
print("OK: /resume lista as conversas salvas (as da janela) e carrega a escolhida")
print("\nTUDO OK.")