george-claudeandClaude Opus 5.5
0b70aad64a
Une entrée user arrivée après la fin d'un tour de l'agent (dernière réponse du fil principal avec stop_reason autre que tool_use) ouvre une nouvelle plage, même écrite par Claude Code (isMeta) : message ou avis d'une autre session, commande slash, renommage. Une entrée isMeta écrite pendant un tour, ou après la fin d'un tour de sous-agent, ne coupe rien. Ajoute le fichier VERSION (1.0.1), vérifié égal à la constante de l'outil. Refs #2 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KT2yBsd7nPivLnrEtcvuhX
259 lines
12 KiB
Python
259 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Test de outils/temps-session.py, hors ligne.
|
|
|
|
python3 -I tests/test-temps-session.py
|
|
|
|
Un transcript fabriqué dans un HOME jetable, et un Gitea simulé (l'appel
|
|
réseau est remplacé) : on vérifie ce que l'outil propose de saisir.
|
|
"""
|
|
import contextlib, datetime, importlib.util, io, json, os, sys, tempfile
|
|
from pathlib import Path
|
|
|
|
DEPOT = Path(__file__).resolve().parent.parent
|
|
OUTIL = DEPOT / "outils" / "temps-session.py"
|
|
SESSION = "00000000-0000-0000-0000-000000000000"
|
|
echecs = []
|
|
|
|
|
|
def verifier(condition, message):
|
|
if not condition:
|
|
echecs.append(message)
|
|
|
|
|
|
def charger():
|
|
spec = importlib.util.spec_from_file_location("temps_session", OUTIL)
|
|
m = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(m)
|
|
return m
|
|
|
|
|
|
def humain(t, texte):
|
|
return {"type": "user", "timestamp": t, "message": {"role": "user", "content": texte}}
|
|
|
|
|
|
def agent(t, fin=None):
|
|
"""Réponse de l'agent ; fin = son stop_reason (end_turn : il rend la main)."""
|
|
return {"type": "assistant", "timestamp": t, "message": {"role": "assistant",
|
|
"content": [{"type": "text", "text": "…"}], "stop_reason": fin}}
|
|
|
|
|
|
def meta(t, texte):
|
|
"""Entrée user écrite par Claude Code, pas par l'humain : message d'une
|
|
autre session, avis, rappel système, sortie de hook."""
|
|
return {"type": "user", "isMeta": True, "timestamp": t,
|
|
"message": {"role": "user", "content": texte}}
|
|
|
|
|
|
def resultat(t):
|
|
return {"type": "user", "timestamp": t, "message": {"role": "user", "content": [
|
|
{"type": "tool_result", "tool_use_id": "x", "content": "…"}]}}
|
|
|
|
|
|
# Deux tours : A de 10:00 à 10:05 (300 s), B de 10:08 à 10:10 (120 s).
|
|
TRANSCRIPT = [humain("2026-10-21T10:00:00Z", "A"), agent("2026-10-21T10:02:00Z"),
|
|
agent("2026-10-21T10:05:00Z"),
|
|
humain("2026-10-21T10:08:00Z", "B"), agent("2026-10-21T10:10:00Z")]
|
|
|
|
|
|
# (api, jeton) de chaque appel au Gitea simulé, pour le dernier lancement.
|
|
APPELS = []
|
|
|
|
|
|
def lancer(ts, saisies, *options, jeton=True, refus=False, session=SESSION):
|
|
"""Lance main() avec un Gitea qui renvoie `saisies` ; rend les lignes JSON.
|
|
jeton=False n'ajoute pas --jeton ; refus=True attend que l'outil refuse,
|
|
sans avoir appelé Gitea."""
|
|
def gitea(api, chemin, j):
|
|
APPELS.append((api, j))
|
|
if chemin == "/user":
|
|
return {"login": "agent"}
|
|
return saisies if chemin.endswith("&page=1") else []
|
|
ts.gitea = gitea
|
|
APPELS.clear()
|
|
sys.argv = ["temps-session.py", "--depot", "organisation/depot", "--ticket", "1",
|
|
"--session", session, *(["--jeton", str(JETON)] if jeton else []), *options]
|
|
sortie = io.StringIO()
|
|
erreur = io.StringIO()
|
|
try:
|
|
with contextlib.redirect_stdout(sortie), contextlib.redirect_stderr(erreur):
|
|
ts.main()
|
|
except SystemExit:
|
|
verifier(refus, "l'outil refuse %s : %s" % (" ".join(options), erreur.getvalue().strip()[-120:]))
|
|
verifier(not APPELS, "l'outil refuse %s après avoir appelé %r" % (" ".join(options), APPELS))
|
|
return []
|
|
verifier(not refus, "l'outil accepte %s" % " ".join(options))
|
|
return [json.loads(l) for l in sortie.getvalue().splitlines() if l.strip()]
|
|
|
|
|
|
with tempfile.TemporaryDirectory() as t:
|
|
t = Path(t)
|
|
os.environ["HOME"] = str(t)
|
|
projet = t / ".claude" / "projects" / "-espace"
|
|
projet.mkdir(parents=True)
|
|
(projet / (SESSION + ".jsonl")).write_text("".join(json.dumps(d) + "\n" for d in TRANSCRIPT))
|
|
for v in ("XDG_CONFIG_HOME", "TKFA_CONFIG", "GITEA_TOKEN_FILE"):
|
|
os.environ.pop(v, None)
|
|
JETON = t / "jeton"
|
|
JETON.write_text("factice\n")
|
|
ts = charger()
|
|
|
|
# Configuration des essais : `autre` par défaut (son jeton, chemin relatif
|
|
# au fichier), `variable` (jeton par GITEA_TOKEN_FILE), et deux instances
|
|
# à refuser. Les essais 1 à 7 passent --jeton, qui prime.
|
|
conf = t / "conf"
|
|
conf.mkdir()
|
|
(conf / "autre.jeton").write_text("autre\n")
|
|
config = conf / "tkfa.toml"
|
|
config.write_text('''defaut = "autre"
|
|
[instances.autre]
|
|
api = "https://gitea.example.org/api/v1/"
|
|
jeton = "autre.jeton"
|
|
[instances.variable]
|
|
api = "https://gitea.example.com/api/v1"
|
|
jeton_env = "GITEA_TOKEN_FILE"
|
|
[instances.sans-jeton]
|
|
api = "https://gitea.example.net/api/v1"
|
|
[instances.en-clair]
|
|
api = "http://gitea.example.org/api/v1"
|
|
jeton = "autre.jeton"
|
|
''')
|
|
os.environ["TKFA_CONFIG"] = str(config)
|
|
|
|
# 1. Rien de saisi : une séance (tours à moins de --seance), 420 s, datée de sa fin.
|
|
lignes = lancer(ts, [])
|
|
verifier([l["time"] for l in lignes] == [420], "sans saisie : %r au lieu de 420 s" % lignes)
|
|
verifier(lignes and lignes[0]["created"].startswith("2026-10-21T10:10:00"),
|
|
"la saisie n'est pas datée de la fin de la séance : %r" % lignes)
|
|
|
|
# 2. A saisi, Gitea a ignoré la date envoyée et daté la saisie de son envoi
|
|
# (10:06, après A, avant B) : seul B reste, 120 s.
|
|
lignes = lancer(ts, [{"created": "2026-10-21T10:06:00Z", "time": 300}])
|
|
verifier([l["time"] for l in lignes] == [120],
|
|
"saisie datée de son envoi : %r au lieu de 120 s (le temps de B est perdu)" % lignes)
|
|
|
|
# 3. A saisi, Gitea a gardé la date envoyée (la fin de A) : seul B reste.
|
|
lignes = lancer(ts, [{"created": "2026-10-21T10:05:00Z", "time": 300}])
|
|
verifier([l["time"] for l in lignes] == [120], "saisie datée de la fin de A : %r au lieu de 120 s" % lignes)
|
|
|
|
# 4. Tout saisi : relancer ne double rien.
|
|
lignes = lancer(ts, [{"created": "2026-10-21T10:06:00Z", "time": 300},
|
|
{"created": "2026-10-21T10:11:00Z", "time": 120}])
|
|
verifier(lignes == [], "tout est saisi, et l'outil propose encore %r" % lignes)
|
|
|
|
# 5. --jusqua : le travail d'après la borne (B, sur un autre ticket) n'est pas compté.
|
|
lignes = lancer(ts, [], "--jusqua", "2026-10-21T10:07:00Z")
|
|
verifier([l["time"] for l in lignes] == [300], "--jusqua entre A et B : %r au lieu de 300 s" % lignes)
|
|
verifier(lignes and lignes[0]["created"].startswith("2026-10-21T10:05:00"),
|
|
"--jusqua : la saisie n'est pas datée de la fin de A : %r" % lignes)
|
|
|
|
# 6. --jusqua au milieu d'un tour : seul ce qui précède la borne compte.
|
|
lignes = lancer(ts, [], "--jusqua", "2026-10-21T10:03:00Z")
|
|
verifier([l["time"] for l in lignes] == [120], "--jusqua dans A : %r au lieu de 120 s" % lignes)
|
|
|
|
# 7. --depuis et --jusqua ensemble, sans fuseau (celui de la machine) : B seul.
|
|
def local(h):
|
|
return datetime.datetime.fromisoformat(h).astimezone().replace(tzinfo=None).isoformat()
|
|
lignes = lancer(ts, [], "--depuis", local("2026-10-21T10:07:00+00:00"),
|
|
"--jusqua", local("2026-10-21T10:30:00+00:00"))
|
|
verifier([l["time"] for l in lignes] == [120], "--depuis/--jusqua sans fuseau : %r au lieu de 120 s" % lignes)
|
|
|
|
# 8. Sans configuration, aucune instance : refus, même avec --jeton et
|
|
# GITEA_TOKEN_FILE.
|
|
del os.environ["TKFA_CONFIG"]
|
|
os.environ["GITEA_TOKEN_FILE"] = str(JETON)
|
|
lancer(ts, [], refus=True)
|
|
lancer(ts, [], jeton=False, refus=True)
|
|
|
|
# 9. L'instance par défaut : son api, son jeton, et pas GITEA_TOKEN_FILE,
|
|
# resté défini ; `jeton_env` sert l'instance qui le nomme.
|
|
lancer(ts, [], "--config", str(config), jeton=False)
|
|
verifier(APPELS and all(a == ("https://gitea.example.org/api/v1", "autre") for a in APPELS),
|
|
"instance par défaut : appels %r" % APPELS)
|
|
lancer(ts, [], "--config", str(config), "--instance", "variable", jeton=False)
|
|
verifier(APPELS and all(a == ("https://gitea.example.com/api/v1", "factice") for a in APPELS),
|
|
"jeton_env : appels %r" % APPELS)
|
|
|
|
# 10. TKFA_CONFIG, puis le chemin par défaut, sont lus aussi.
|
|
os.environ["TKFA_CONFIG"] = str(config)
|
|
lancer(ts, [], jeton=False)
|
|
verifier(APPELS and APPELS[0][0] == "https://gitea.example.org/api/v1", "TKFA_CONFIG : appels %r" % APPELS)
|
|
del os.environ["TKFA_CONFIG"]
|
|
(t / ".config" / "tkfa").mkdir(parents=True)
|
|
(t / ".config" / "tkfa" / "config.toml").write_text(config.read_text().replace(
|
|
'"autre.jeton"', '"%s"' % (conf / "autre.jeton")))
|
|
lancer(ts, [], jeton=False)
|
|
verifier(APPELS and APPELS[0] == ("https://gitea.example.org/api/v1", "autre"),
|
|
"~/.config/tkfa/config.toml : appels %r" % APPELS)
|
|
(t / ".config" / "tkfa" / "config.toml").unlink()
|
|
|
|
# 11. Refus, avant tout appel : instance sans jeton (GITEA_TOKEN_FILE est
|
|
# défini mais ne part pas ailleurs), variable de `jeton_env` absente,
|
|
# api en http, instance ou fichier absents.
|
|
lancer(ts, [], "--config", str(config), "--instance", "sans-jeton", jeton=False, refus=True)
|
|
del os.environ["GITEA_TOKEN_FILE"]
|
|
lancer(ts, [], "--config", str(config), "--instance", "variable", jeton=False, refus=True)
|
|
lancer(ts, [], "--config", str(config), "--instance", "en-clair", refus=True)
|
|
lancer(ts, [], "--config", str(config), "--instance", "inconnue", refus=True)
|
|
lancer(ts, [], "--config", str(t / "absent.toml"), refus=True)
|
|
|
|
# 12. --version : le nom de l'outil et sa version, sans rien lire d'autre.
|
|
sortie = io.StringIO()
|
|
sys.argv = ["temps-session.py", "--version"]
|
|
try:
|
|
with contextlib.redirect_stdout(sortie):
|
|
ts.main()
|
|
except SystemExit:
|
|
pass
|
|
verifier(sortie.getvalue().strip() == "temps-session.py " + ts.VERSION,
|
|
"--version affiche %r" % sortie.getvalue())
|
|
|
|
# 13. Le fichier VERSION et la constante de l'outil (qui se télécharge
|
|
# seul) disent la même version.
|
|
version = (DEPOT / "VERSION").read_text().strip() if (DEPOT / "VERSION").exists() else None
|
|
verifier(version == ts.VERSION, "VERSION dit %r, l'outil %r" % (version, ts.VERSION))
|
|
|
|
# 14 à 17 : entrées user écrites par Claude Code (isMeta). Chaque cas a
|
|
# sa session ; le tour A va de 10:00 à 10:05 (300 s).
|
|
def cas(numero, entrees):
|
|
session = "00000000-0000-0000-0000-0000000000%02d" % numero
|
|
(projet / (session + ".jsonl")).write_text("".join(json.dumps(d) + "\n" for d in entrees))
|
|
return [l["time"] for l in lancer(ts, [], "--config", str(config), session=session)]
|
|
|
|
tour_a = [humain("2026-10-21T10:00:00Z", "A"), agent("2026-10-21T10:05:00Z", "end_turn")]
|
|
reprise = [agent("2026-10-21T10:26:00Z", "tool_use"), resultat("2026-10-21T10:26:30Z"),
|
|
agent("2026-10-21T10:27:00Z", "end_turn")]
|
|
|
|
# 14. L'agent a rendu la main ; une autre session lui écrit 20 min plus
|
|
# tard (sous --pause) et il reprend : l'attente n'est pas comptée,
|
|
# restent A et la reprise (10:25 à 10:27), 420 s.
|
|
temps = cas(14, tour_a + [meta("2026-10-21T10:25:00Z", "Another Claude session sent a message: "
|
|
'<cross-session-message from="autre">…</cross-session-message>')] + reprise)
|
|
verifier(temps == [420], "message d'une autre session après l'attente : %r au lieu de 420 s" % temps)
|
|
|
|
# 15. Même chose pour un avis d'inactivité d'une autre session.
|
|
temps = cas(15, tour_a + [meta("2026-10-21T10:25:00Z", '[Cross-session idle notice] "autre" …')] + reprise)
|
|
verifier(temps == [420], "avis d'une autre session après l'attente : %r au lieu de 420 s" % temps)
|
|
|
|
# 16. Une entrée isMeta écrite pendant un tour (rappel système, sortie de
|
|
# hook) ne coupe pas la plage : 300 s, pas 240.
|
|
temps = cas(16, [humain("2026-10-21T10:00:00Z", "A"), agent("2026-10-21T10:02:00Z", "tool_use"),
|
|
meta("2026-10-21T10:03:00Z", "<system-reminder>…</system-reminder>"),
|
|
resultat("2026-10-21T10:04:00Z"), agent("2026-10-21T10:05:00Z", "end_turn")])
|
|
verifier(temps == [300], "entrée isMeta pendant un tour : %r au lieu de 300 s" % temps)
|
|
|
|
# 17. Un sous-agent qui finit son tour (isSidechain) ne rend pas la main
|
|
# de l'agent principal : la plage n'est pas coupée non plus.
|
|
fin_sous_agent = agent("2026-10-21T10:02:30Z", "end_turn")
|
|
fin_sous_agent["isSidechain"] = True
|
|
temps = cas(17, [humain("2026-10-21T10:00:00Z", "A"), agent("2026-10-21T10:02:00Z", "tool_use"),
|
|
fin_sous_agent, meta("2026-10-21T10:03:00Z", "<system-reminder>…</system-reminder>"),
|
|
resultat("2026-10-21T10:04:00Z"), agent("2026-10-21T10:05:00Z", "end_turn")])
|
|
verifier(temps == [300], "fin de tour d'un sous-agent : %r au lieu de 300 s" % temps)
|
|
|
|
if echecs:
|
|
print("%d échec(s) :" % len(echecs))
|
|
for e in echecs:
|
|
print(" ✗", e)
|
|
sys.exit(1)
|
|
print("ok : outil de saisie des temps")
|