fix(cron): forward attach_to_session through cronjob handler (salvage #84829)

contributors/emails/kshitijkapoorr@gmail.com

@@ -0,0 +1 @@
+kshitijk4poor

tests/tools/test_cronjob_tools.py

@@ -444,6 +444,123 @@ def test_handler_update_leaves_user_pin_untouched(self):
         assert stored["name"] == "renamed"
 
 
+class TestRegisteredHandlerForwardsAttachToSession:
+    """#84802 — schema + cronjob() already accept attach_to_session, but the
+    registry adapter must forward it or create silently drops the field and
+    update returns "No updates provided." """
+
+    @pytest.fixture(autouse=True)
+    def _setup_cron_dir(self, tmp_path, monkeypatch):
+        monkeypatch.setattr("cron.jobs.CRON_DIR", tmp_path / "cron")
+        monkeypatch.setattr("cron.jobs.JOBS_FILE", tmp_path / "cron" / "jobs.json")
+        monkeypatch.setattr("cron.jobs.OUTPUT_DIR", tmp_path / "cron" / "output")
+
+    def test_create_persists_attach_to_session(self):
+        from cron.jobs import get_job
+        from tools.registry import registry
+
+        created = json.loads(
+            registry.dispatch(
+                "cronjob",
+                {
+                    "action": "create",
+                    "name": "Continuable cron canary",
+                    "schedule": "1h",
+                    "repeat": 1,
+                    "deliver": "origin",
+                    "attach_to_session": True,
+                    "prompt": "Reply exactly: canary",
+                },
+            )
+        )
+        assert created["success"] is True
+        assert created.get("job", {}).get("attach_to_session") is True
+        stored = get_job(created["job_id"])
+        assert stored is not None
+        assert stored.get("attach_to_session") is True
+        listing = json.loads(registry.dispatch("cronjob", {"action": "list"}))
+        listed = next(j for j in listing["jobs"] if j["job_id"] == created["job_id"])
+        assert listed.get("attach_to_session") is True
+
+    def test_update_persists_attach_to_session(self):
+        from cron.jobs import get_job
+        from tools.registry import registry
+
+        created = json.loads(
+            registry.dispatch(
+                "cronjob",
+                {
+                    "action": "create",
+                    "name": "plain",
+                    "schedule": "1h",
+                    "prompt": "Reply exactly: canary",
+                },
+            )
+        )
+        assert created["success"] is True
+        assert "attach_to_session" not in (get_job(created["job_id"]) or {})
+
+        updated = json.loads(
+            registry.dispatch(
+                "cronjob",
+                {
+                    "action": "update",
+                    "job_id": created["job_id"],
+                    "attach_to_session": True,
+                },
+            )
+        )
+        assert updated["success"] is True, updated
+        assert updated.get("job", {}).get("attach_to_session") is True
+        stored = get_job(created["job_id"])
+        assert stored is not None
+        assert stored.get("attach_to_session") is True
+
+        disabled = json.loads(
+            registry.dispatch(
+                "cronjob",
+                {
+                    "action": "update",
+                    "job_id": created["job_id"],
+                    "attach_to_session": False,
+                },
+            )
+        )
+        assert disabled["success"] is True, disabled
+        assert disabled.get("job", {}).get("attach_to_session") is False
+        stored = get_job(created["job_id"])
+        assert stored is not None
+        assert stored.get("attach_to_session") is False
+        listing = json.loads(registry.dispatch("cronjob", {"action": "list"}))
+        listed = next(j for j in listing["jobs"] if j["job_id"] == created["job_id"])
+        assert listed.get("attach_to_session") is False
+
+    def test_omitted_create_leaves_field_absent(self):
+        from cron.jobs import get_job
+        from tools.registry import registry
+
+        created = json.loads(
+            registry.dispatch(
+                "cronjob",
+                {
+                    "action": "create",
+                    "schedule": "1h",
+                    "prompt": "fire and forget",
+                },
+            )
+        )
+        assert created["success"] is True
+        stored = get_job(created["job_id"])
+        assert stored is not None
+        assert "attach_to_session" not in stored
+        # And the formatted list output must not invent the field either.
+        listed = json.loads(registry.dispatch("cronjob", {"action": "list"}))
+        formatted = next(
+            j for j in listed["jobs"] if j["job_id"] == created["job_id"]
+        )
+        assert "attach_to_session" not in formatted
+
+
 class TestLocalDeliveryNotice:
     """#51568 — TUI/CLI cron jobs are local-only; surface that at create time
     so the agent doesn't promise a delivery that never happens."""

tools/cronjob_tools.py

@@ -804,6 +804,8 @@ def _format_job(job: Dict[str, Any]) -> Dict[str, Any]:
     ]
     if external_refs:
         result["context_from"] = external_refs
+    if isinstance(job.get("attach_to_session"), bool):
+        result["attach_to_session"] = job["attach_to_session"]
     return result
 
 
@@ -1970,6 +1972,7 @@ def _cronjob_handler(args, **kw):
         enabled_toolsets=args.get("enabled_toolsets"),
         workdir=args.get("workdir"),
         no_agent=args.get("no_agent"),
+        attach_to_session=args.get("attach_to_session"),
         monitor_script=_mon_script,
         monitor_url=_mon_url,
         task_id=kw.get("task_id"),