Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
135 changes: 126 additions & 9 deletions backend/app/api/endpoints/deliveries.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,7 @@
ProjectChatAgent,
loop_datetime_is_unset,
)
from app.models.project_chat_message import ProjectChatMessage
from app.models.user import User
from app.schemas.base_role import BaseRole
from app.schemas.delivery import (
Expand Down Expand Up @@ -64,6 +65,12 @@
MyWorkItemResponse,
MyWorkListResponse,
)
from app.schemas.human_issue_work import (
HumanWorkActionResponse,
HumanWorkReview,
HumanWorkStart,
HumanWorkSubmit,
)
from app.schemas.issue_assignment import (
IssueAssignmentCreate,
IssueAssignmentCreateResponse,
Expand All @@ -88,6 +95,7 @@
require_issue_action,
)
from app.services.delivery import delivery_service
from app.services.human_issue_work import human_issue_work_service
from app.services.issue_assignments import issue_assignment_service
from app.services.issue_workflow_decision import issue_workflow_decision_service
from app.services.issue_workflow_planning import issue_workflow_planning_service
Expand All @@ -96,7 +104,9 @@
from app.services.loop_item_status_history import (
is_processing_status,
project_status_transition,
write_status_change,
)
from app.services.loop_item_unread import advance_content_revision
from app.services.loop_items import MY_WORK_ITEM_LIMIT, loop_item_service
from app.services.loop_items.external_provider import external_loop_item_provider
from app.services.loop_items.provider_router import (
Expand Down Expand Up @@ -135,6 +145,81 @@ def _loop_item_response(
)


async def _human_work_response(
db: Session,
item: LoopItem,
user: User,
message: ProjectChatMessage | None = None,
*,
created: bool = False,
) -> HumanWorkActionResponse:
from app.api.ws.wework_runtime_namespace import (
PROJECT_CHAT_CREATED_EVENT,
WEWORK_RUNTIME_NAMESPACE,
project_chat_room,
)
from app.core.socketio import get_sio
from app.services.project_chat.service import ProjectChatService

message_view = ProjectChatService.to_view(message) if message is not None else None
if created and message_view is not None:
try:
await get_sio().emit(
PROJECT_CHAT_CREATED_EVENT,
message_view.model_dump(mode="json", by_alias=True),
room=project_chat_room(str(item.cloud_project_id), item.id),
namespace=WEWORK_RUNTIME_NAMESPACE,
)
except Exception:
logger.exception(
"Human Issue activity broadcast failed: item_id=%s", item.id
)
return HumanWorkActionResponse(
issue=_loop_item_response(db, item, user), message=message_view
)


@router.post("/loop-items/{item_id}/work/start", response_model=HumanWorkActionResponse)
async def start_human_issue_work(
item_id: str,
values: HumanWorkStart,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> HumanWorkActionResponse:
item = human_issue_work_service.start(db, item_id, current_user.id, values)
return await _human_work_response(db, item, current_user)


@router.post(
"/loop-items/{item_id}/work/submit", response_model=HumanWorkActionResponse
)
async def submit_human_issue_work(
item_id: str,
values: HumanWorkSubmit,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> HumanWorkActionResponse:
item, message, created = human_issue_work_service.submit(
db, item_id, current_user.id, values
)
return await _human_work_response(db, item, current_user, message, created=created)


@router.post(
"/loop-items/{item_id}/work/review", response_model=HumanWorkActionResponse
)
async def review_human_issue_work(
item_id: str,
values: HumanWorkReview,
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> HumanWorkActionResponse:
item, message, created = human_issue_work_service.review(
db, item_id, current_user.id, values
)
return await _human_work_response(db, item, current_user, message, created=created)


def _automation_selection_error(
*,
code: str,
Expand Down Expand Up @@ -573,6 +658,7 @@ async def create_loop_item(
)
project_workflow = project_metadata.get("workflow_definition")
explicit_workflow = values.workflow
has_explicit_human_assignee = values.assignee_user_id is not None
has_bound_workflow = (
explicit_workflow is not None
and (
Expand All @@ -586,7 +672,9 @@ async def create_loop_item(
or project_workflow.get("advancement_policy") == "ai"
)
)
if has_bound_workflow:
if has_explicit_human_assignee:
selected_automation_id = None
elif has_bound_workflow:
if values.automation_rule_id:
raise _automation_selection_error(
code="automation_selection_stale",
Expand Down Expand Up @@ -627,9 +715,12 @@ async def create_loop_item(
)
if created.internal_item is not None:
db.refresh(created.internal_item)
if issue_workflow_start_service.should_start_after_creation(
created.internal_item,
project,
if (
not has_explicit_human_assignee
and issue_workflow_start_service.should_start_after_creation(
created.internal_item,
project,
)
):
await issue_workflow_start_service.start(
db,
Expand Down Expand Up @@ -1158,9 +1249,13 @@ async def update_loop_item(
current_status=item.status,
)
entered_processing = status_changed and status_transition.entered_processing
should_start_workflow = selected_status_automation_id is None and (
entered_processing
or (workflow_updated and is_processing_status(project, item.status))
should_start_workflow = (
not human_issue_work_service.is_direct_human_assignment(db, item)
and selected_status_automation_id is None
and (
entered_processing
or (workflow_updated and is_processing_status(project, item.status))
)
)
logger.info(
"[issue-workflow-start] update item=%s project=%s previous_status=%s "
Expand Down Expand Up @@ -1446,7 +1541,7 @@ async def create_issue_assignment(
member_id=values.target_id,
workflow_step=values.workflow_step,
)
if existing is not None:
if existing is not None and existing.metadata.get("trigger") != "default":
raise HTTPException(
status.HTTP_409_CONFLICT,
"Assignment is already active for this workflow step",
Expand Down Expand Up @@ -1593,15 +1688,37 @@ def remove_issue_assignment(
issue_id=item_id,
user_id=current_user.id,
)
issue_assignment_service.remove(
was_human = human_issue_work_service.is_direct_human_assignment(db, item)
removed = issue_assignment_service.remove(
db,
project_id=int(project.id),
issue_id=item_id,
assignment_id=assignment_id,
user_id=current_user.id,
)
issue_assignment_service.project_legacy_assignment(db, item=item)
if was_human and removed.member_type == "human":
metadata = dict(item.metadata_json or {})
metadata.pop("human_work", None)
if item.status in {"in_progress", "in_review"}:
write_status_change(
metadata,
project=project,
from_status=item.status,
to_status="pending",
trigger="unassigned",
by_user_id=current_user.id,
)
item.status = "pending"
item.sort_order = 0
item.completed_at = None
item.metadata_json = advance_content_revision(
metadata, actor_user_id=current_user.id
)
db.commit()
publish_loop_item_changed(
db, item=item, reason="assignment_removed", actor_user_id=current_user.id
)


@router.get(
Expand Down
13 changes: 13 additions & 0 deletions backend/app/api/ws/device_namespace.py
Original file line number Diff line number Diff line change
Expand Up @@ -1015,6 +1015,19 @@ def _project_bound_runtime_event_status(
return None
binding_metadata["runtime_status_event_seq"] = event_seq
binding.metadata_json = binding_metadata
from app.services.human_issue_work import human_issue_work_service

if human_issue_work_service.is_direct_human_assignment(db, item_before):
logger.info(
"[IssueTaskRuntimeSync] kept human Issue status "
"user=%s device=%s task=%s event=%s item=%s",
user_id,
device_id,
task_id,
event_name,
item_before.id,
)
return None
next_status = (
"in_progress"
if projected_status == "running"
Expand Down
1 change: 1 addition & 0 deletions backend/app/schemas/delivery.py
Original file line number Diff line number Diff line change
Expand Up @@ -140,6 +140,7 @@ class LoopItemResponse(BaseModel):
assignment_history: list[dict[str, Any]] = Field(default_factory=list)
status_history: list[dict[str, Any]] = Field(default_factory=list)
approval: dict[str, Any] | None = None
human_work: dict[str, Any] | None = None
queued_at: str | None = None
execution_note: str | None = None
execution_error: str | None = None
Expand Down
37 changes: 37 additions & 0 deletions backend/app/schemas/human_issue_work.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,37 @@
# SPDX-FileCopyrightText: 2026 Weibo, Inc.
# SPDX-License-Identifier: Apache-2.0

"""Versioned actions for a human assigned to a native cloud Issue."""

from typing import Literal

from pydantic import BaseModel, Field, model_validator

from app.schemas.delivery import LoopItemResponse
from app.schemas.project_chat import ProjectChatMessageView


class HumanWorkStart(BaseModel):
version: int = Field(ge=1)


class HumanWorkSubmit(HumanWorkStart):
request_id: str = Field(min_length=1, max_length=64)
summary: str = Field(min_length=1, max_length=100_000)


class HumanWorkReview(HumanWorkStart):
request_id: str = Field(min_length=1, max_length=64)
decision: Literal["accept", "request_changes"]
reason: str | None = Field(default=None, max_length=10_000)

@model_validator(mode="after")
def require_rejection_reason(self) -> "HumanWorkReview":
if self.decision == "request_changes" and not (self.reason or "").strip():
raise ValueError("A reason is required when requesting changes")
return self


class HumanWorkActionResponse(BaseModel):
issue: LoopItemResponse
message: ProjectChatMessageView | None = None
20 changes: 11 additions & 9 deletions backend/app/schemas/workspace.py
Original file line number Diff line number Diff line change
Expand Up @@ -88,15 +88,17 @@ def validate_repositories(self) -> "ExecutionEnvironmentDefinition":
for right in paths[index + 1 :]
):
raise ValueError("repository paths must not overlap")
for step in self.setup_steps:
working_directory = step.working_directory
if working_directory and not any(
working_directory == path or working_directory.startswith(f"{path}/")
for path in paths
):
raise ValueError(
"setup working directory must be inside a configured repository"
)
if paths:
for step in self.setup_steps:
working_directory = step.working_directory
if working_directory and not any(
working_directory == path
or working_directory.startswith(f"{path}/")
for path in paths
):
raise ValueError(
"setup working directory must be inside a configured repository"
)
return self


Expand Down
17 changes: 17 additions & 0 deletions backend/app/services/delivery/service.py
Original file line number Diff line number Diff line change
Expand Up @@ -387,6 +387,23 @@ def finalize(
actor_user_id=user_id,
)
return delivery
from app.services.human_issue_work import human_issue_work_service

if human_issue_work_service.is_direct_human_assignment(db, item):
item.current_delivery_id = delivery.id
item.metadata_json = advance_content_revision(
item.metadata_json, actor_user_id=user_id
)
item.version += 1
db.commit()
db.refresh(delivery)
publish_loop_item_changed(
db,
item=item,
reason="delivery_finalized",
actor_user_id=user_id,
)
return delivery
previous_status = item.status
if previous_status != "completed":
project = db.get(CloudProject, item.cloud_project_id)
Expand Down
12 changes: 7 additions & 5 deletions backend/app/services/execution_environment_initialization.py
Original file line number Diff line number Diff line change
Expand Up @@ -143,15 +143,16 @@ async def initialize_execution_environment(
state via :func:`merge_execution_environment_device_state`; the route id is
``runtime_device_route_id(device)``.
"""
# Saving tolerates an empty repository list so a draft environment can be
# stored, but preparation on a device cannot run without one; reject here
# instead of letting the executor surface a raw internal error.
repositories = [
repository
for repository in definition.get("repositories") or []
if isinstance(repository, dict)
]
if sum(1 for repository in repositories if bool(repository.get("primary"))) != 1:
if (
repositories
and sum(1 for repository in repositories if bool(repository.get("primary")))
!= 1
):
raise HTTPException(
status.HTTP_422_UNPROCESSABLE_CONTENT,
"Execution environment requires exactly one primary repository",
Expand All @@ -177,7 +178,8 @@ async def initialize_execution_environment(
"on this device",
)
try:
await _sync_device_git_credentials(db, device)
if repositories:
await _sync_device_git_credentials(db, device)
result = await execute_configured_device_command(
db=db,
user_id=int(device.user_id),
Expand Down
Loading
Loading