# -*- coding: utf-8 -*- """Skill Activity Stream: journal, flush, job_id, finish stdout, truncation.""" from __future__ import annotations import io import json import os import shutil import tempfile import unittest from pathlib import Path from unittest.mock import patch from jiangchang_skill_core import activity from jiangchang_skill_core.runtime_env import get_runs_dir class _ActivityEnvIsolation(unittest.TestCase): _KEYS = ( "JIANGCHANG_DATA_ROOT", "JIANGCHANG_USER_ID", "JIANGCHANG_JOB_ID", "JIANGCHANG_RUN_MODE", "JIANGCHANG_ACTIVITY_DEBUG", ) def setUp(self) -> None: self._saved = {k: os.environ.get(k) for k in self._KEYS} self._tmps: list[str] = [] for k in self._KEYS: os.environ.pop(k, None) activity._reset_for_tests() def tearDown(self) -> None: activity._reset_for_tests() for k, v in self._saved.items(): if v is None: os.environ.pop(k, None) else: os.environ[k] = v for tmp in self._tmps: shutil.rmtree(tmp, ignore_errors=True) def _tmp_data_root(self) -> str: tmp = tempfile.mkdtemp() self._tmps.append(tmp) return tmp class TestActivityJournal(_ActivityEnvIsolation): def test_emit_writes_ndjson_with_meta(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "job-test-1" activity._reset_for_tests() activity.emit("步骤一", type="activity", skill="geo-batch") journal = Path(get_runs_dir()) / "job-test-1.jsonl" self.assertTrue(journal.is_file()) lines = journal.read_text(encoding="utf-8").strip().splitlines() self.assertGreaterEqual(len(lines), 2) meta = json.loads(lines[0]) self.assertEqual(meta["type"], "meta") self.assertEqual(meta["job_id"], "job-test-1") event = json.loads(lines[1]) self.assertEqual(event["type"], "activity") self.assertEqual(event["text"], "步骤一") self.assertEqual(event["schema_version"], 1) self.assertEqual(event["skill"], "geo-batch") self.assertIn("ts", event) def test_emit_flushes_immediately(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "job-flush" activity._reset_for_tests() activity.emit("flush test") journal = Path(get_runs_dir()) / "job-flush.jsonl" self.assertGreater(journal.stat().st_size, 0) def test_job_id_from_env(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "env-job-42" activity._reset_for_tests() self.assertEqual(activity.get_job_id(), "env-job-42") def test_job_id_generated_and_cached(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp activity._reset_for_tests() jid1 = activity.get_job_id() jid2 = activity.get_job_id() self.assertEqual(jid1, jid2) self.assertEqual(len(jid1), 32) def test_debug_skipped_without_env(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "job-no-debug" activity._reset_for_tests() activity.emit("hidden", type="debug") journal = Path(get_runs_dir()) / "job-no-debug.jsonl" self.assertFalse(journal.exists()) def test_debug_written_when_enabled(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "job-debug" os.environ["JIANGCHANG_ACTIVITY_DEBUG"] = "1" activity._reset_for_tests() activity.emit("dbg", type="debug") journal = Path(get_runs_dir()) / "job-debug.jsonl" content = journal.read_text(encoding="utf-8") self.assertIn('"type":"debug"', content.replace(" ", "")) def test_emit_never_prints_stdout(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "job-quiet" activity._reset_for_tests() buf = io.StringIO() with patch("sys.stdout", buf): activity.emit("silent step") self.assertEqual(buf.getvalue(), "") def test_get_run_mode_default_cli(self) -> None: self.assertEqual(activity.get_run_mode(), "cli") def test_get_run_mode_job(self) -> None: os.environ["JIANGCHANG_RUN_MODE"] = "job" self.assertEqual(activity.get_run_mode(), "job") class TestActivityFinish(_ActivityEnvIsolation): def test_finish_stdout_single_line_result(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "job-finish" activity._reset_for_tests() buf = io.StringIO() with patch("sys.stdout", buf): activity.finish(status="success", message="ok", skill="geo", generated=3) lines = [ln for ln in buf.getvalue().splitlines() if ln.strip()] self.assertEqual(len(lines), 1) result = json.loads(lines[0]) self.assertEqual(result["type"], "result") self.assertEqual(result["schema_version"], 1) self.assertEqual(result["status"], "success") self.assertEqual(result["skill"], "geo") self.assertEqual(result["generated"], 3) journal = Path(get_runs_dir()) / "job-finish.jsonl" journal_lines = journal.read_text(encoding="utf-8").strip().splitlines() last = json.loads(journal_lines[-1]) self.assertEqual(last["type"], "done") self.assertEqual(last["status"], "success") def test_finish_failed_writes_error_event(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "job-fail" activity._reset_for_tests() with patch("sys.stdout", io.StringIO()): activity.finish(status="failed", message="boom") journal = Path(get_runs_dir()) / "job-fail.jsonl" last = json.loads(journal.read_text(encoding="utf-8").strip().splitlines()[-1]) self.assertEqual(last["type"], "error") def test_finish_truncates_large_result(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp os.environ["JIANGCHANG_JOB_ID"] = "job-trunc" activity._reset_for_tests() big = {f"field_{i}": "x" * 200 for i in range(50)} buf = io.StringIO() with patch("sys.stdout", buf): activity.finish(status="success", skill="big", **big) line = buf.getvalue().strip() self.assertLessEqual(len(line), 4096) result = json.loads(line) self.assertTrue(result.get("truncated")) class TestJobContext(_ActivityEnvIsolation): def test_job_context_auto_finish_on_exception(self) -> None: tmp = self._tmp_data_root() os.environ["JIANGCHANG_DATA_ROOT"] = tmp activity._reset_for_tests() with patch("sys.stdout", io.StringIO()): with self.assertRaises(ValueError): with activity.job_context(job_id="ctx-err", skill="test-skill"): raise ValueError("broken") journal = Path(get_runs_dir()) / "ctx-err.jsonl" last = json.loads(journal.read_text(encoding="utf-8").strip().splitlines()[-1]) self.assertEqual(last["type"], "error") self.assertIn("broken", last["message"]) if __name__ == "__main__": unittest.main()