mirror of
https://github.com/eigent-ai/eigent.git
synced 2026-08-30 03:11:42 +00:00
274 lines
9.5 KiB
Python
274 lines
9.5 KiB
Python
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
# ========= Copyright 2025-2026 @ Eigent.ai All Rights Reserved. =========
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.controller.run_controller import (
|
|
CancelRunBody,
|
|
ForkRunBody,
|
|
InteractionDecisionBody,
|
|
ResumeRunBody,
|
|
RunSignalBody,
|
|
cancel_run,
|
|
decide_run_interaction,
|
|
fork_run,
|
|
list_project_runs,
|
|
list_run_interactions,
|
|
resume_run,
|
|
signal_run,
|
|
)
|
|
from app.run_journal import SQLiteRunJournal
|
|
from app.run_runtime import RunCoordinator
|
|
from app.workspace_config.admission import (
|
|
EnvironmentAdmissionService,
|
|
LegacyEnvironmentImporter,
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_control_api_creates_attempt_fork_and_cancel_intent(
|
|
tmp_path,
|
|
):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
journal.ensure_run(run_id="run-1", project_id="project-1")
|
|
journal.reconcile_startup(now=1)
|
|
coordinator = RunCoordinator(journal)
|
|
with patch(
|
|
"app.controller.run_controller.get_default_run_coordinator",
|
|
return_value=coordinator,
|
|
):
|
|
resumed = await resume_run(
|
|
"run-1",
|
|
ResumeRunBody(request_id="resume-1"),
|
|
)
|
|
assert resumed["attempt"]["status"] == "pending"
|
|
assert resumed["execution_state"] == "awaiting_execution_context"
|
|
|
|
forked = await fork_run(
|
|
"run-1",
|
|
ForkRunBody(request_id="fork-1", new_run_id="run-fork"),
|
|
)
|
|
assert forked["run"]["parent_run_id"] == "run-1"
|
|
assert forked["requires_resume"] is True
|
|
|
|
cancelled = await cancel_run(
|
|
"run-1",
|
|
CancelRunBody(request_id="cancel-1"),
|
|
)
|
|
assert cancelled["status"] == "cancelled"
|
|
await coordinator.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resume_control_inherits_latest_environment_binding(tmp_path):
|
|
workspace = tmp_path / "workspace"
|
|
workspace.mkdir()
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
journal.ensure_run(run_id="run-env", project_id="project-1")
|
|
template = LegacyEnvironmentImporter().build_template(
|
|
model_platform="openai",
|
|
model_type="gpt-5",
|
|
auth_source="api_key",
|
|
requested_effort="high",
|
|
allow_local_system=True,
|
|
)
|
|
environment = EnvironmentAdmissionService(journal).persist_for_run(
|
|
run_id="run-env",
|
|
space_id="space-1",
|
|
working_directory=workspace,
|
|
created_by="test-user",
|
|
template=template,
|
|
)
|
|
journal.create_run_attempt(
|
|
"run-env",
|
|
request_id="initial:run-env",
|
|
reason="initial_execution",
|
|
activate=True,
|
|
environment=environment.binding,
|
|
now=1,
|
|
)
|
|
journal.reconcile_startup(now=2)
|
|
|
|
coordinator = RunCoordinator(journal)
|
|
with patch(
|
|
"app.controller.run_controller.get_default_run_coordinator",
|
|
return_value=coordinator,
|
|
):
|
|
resumed = await resume_run(
|
|
"run-env",
|
|
ResumeRunBody(request_id="resume-env"),
|
|
)
|
|
|
|
attempt = resumed["attempt"]
|
|
assert attempt["environment_spec_id"] == environment.spec.spec_id
|
|
assert attempt["environment_spec_digest"] == environment.spec.digest
|
|
assert attempt["thinking_effort_effective"] == "high"
|
|
assert (
|
|
attempt["provider_capability_revision"]
|
|
== environment.spec.provider_capability_revision
|
|
)
|
|
await coordinator.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_signal_api_rejects_cross_run_approval_mutation(tmp_path):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
journal.ensure_run(run_id="run-1", project_id="project-1")
|
|
journal.ensure_run(run_id="run-2", project_id="project-1")
|
|
attempt = journal.create_run_attempt(
|
|
"run-1",
|
|
request_id="initial",
|
|
reason="initial_execution",
|
|
activate=True,
|
|
now=1,
|
|
)
|
|
journal.create_approval(
|
|
approval_id="approval-1",
|
|
run_id="run-1",
|
|
attempt_id=attempt.attempt_id,
|
|
prompt={"question": "Continue?"},
|
|
now=2,
|
|
)
|
|
with patch(
|
|
"app.controller.run_controller.get_default_run_journal",
|
|
return_value=journal,
|
|
):
|
|
with pytest.raises(HTTPException) as error:
|
|
await signal_run(
|
|
"run-2",
|
|
RunSignalBody(
|
|
signal_type="approval.decided",
|
|
signal_id="signal-1",
|
|
payload={
|
|
"approval_id": "approval-1",
|
|
"decision": "approved",
|
|
"expected_version": 0,
|
|
},
|
|
),
|
|
)
|
|
assert error.value.status_code == 409
|
|
assert journal.list_approvals("run-1")[0].status == "pending"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_typed_interaction_api_lists_and_resolves_question(tmp_path):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
task_lock = AsyncMock()
|
|
journal.ensure_run(run_id="run-1", project_id="project-1")
|
|
attempt = journal.create_run_attempt(
|
|
"run-1",
|
|
request_id="initial",
|
|
reason="initial_execution",
|
|
activate=True,
|
|
now=1,
|
|
)
|
|
journal.create_human_interaction(
|
|
interaction_id="question-1",
|
|
run_id="run-1",
|
|
attempt_id=attempt.attempt_id,
|
|
interaction_type="choice",
|
|
request={"question": "Pick one", "agent": "worker"},
|
|
options=[
|
|
{"id": "a", "label": "A"},
|
|
{"id": "b", "label": "B"},
|
|
],
|
|
now=2,
|
|
)
|
|
with (
|
|
patch(
|
|
"app.controller.run_controller.get_default_run_journal",
|
|
return_value=journal,
|
|
),
|
|
patch(
|
|
"app.service.task.get_task_lock_if_exists",
|
|
return_value=task_lock,
|
|
),
|
|
):
|
|
listed = await list_run_interactions("run-1", status="pending")
|
|
resolved = await decide_run_interaction(
|
|
"run-1",
|
|
"question-1",
|
|
InteractionDecisionBody(
|
|
decision_request_id="decision-1",
|
|
decision={"option_id": "b"},
|
|
expected_version=0,
|
|
),
|
|
)
|
|
|
|
assert [item["interaction_id"] for item in listed["interactions"]] == [
|
|
"question-1"
|
|
]
|
|
assert [
|
|
option["option_id"]
|
|
for option in listed["interactions"][0]["options"]
|
|
] == [
|
|
"a",
|
|
"b",
|
|
]
|
|
assert resolved["status"] == "resolved"
|
|
task_lock.put_human_input.assert_awaited_once_with(
|
|
"worker", '{"option_id":"b"}'
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_project_runs_reads_canonical_interrupted_state(tmp_path):
|
|
with SQLiteRunJournal(tmp_path / "journal.sqlite3") as journal:
|
|
journal.ensure_run(run_id="older", project_id="project-1", now=1)
|
|
journal.create_run_attempt(
|
|
"older",
|
|
request_id="initial-older",
|
|
reason="initial_execution",
|
|
activate=True,
|
|
now=1,
|
|
)
|
|
journal.reconcile_startup(now=2)
|
|
journal.ensure_run(run_id="other", project_id="project-2", now=3)
|
|
with (
|
|
patch(
|
|
"app.controller.run_controller.get_default_run_journal",
|
|
return_value=journal,
|
|
),
|
|
patch(
|
|
"app.run_sync.runtime.notify_default_cloud_sync_worker"
|
|
) as notify_sync,
|
|
patch(
|
|
"app.run_sync.runtime."
|
|
"is_default_cloud_history_bootstrap_pending",
|
|
return_value=True,
|
|
),
|
|
patch(
|
|
"app.run_sync.runtime.bootstrap_default_cloud_history",
|
|
new_callable=AsyncMock,
|
|
) as bootstrap_history,
|
|
):
|
|
result = await list_project_runs(
|
|
project_id="project-1",
|
|
status=["interrupted"],
|
|
limit=1,
|
|
)
|
|
|
|
assert [run["run_id"] for run in result["runs"]] == ["older"]
|
|
assert result["runs"][0]["status"] == "interrupted"
|
|
assert result["cloud_restore_pending"] is True
|
|
# Startup recovery must not count the unobserved process-down interval as
|
|
# active execution time. This attempt never persisted a later heartbeat.
|
|
assert result["runs"][0]["total_attempt_elapsed_ms"] == 0
|
|
notify_sync.assert_called_once_with()
|
|
bootstrap_history.assert_not_awaited()
|