diff --git a/backend/alembic/versions/2026_04_24_0000-add_priority_fields_to_tasks.py b/backend/alembic/versions/2026_04_24_0000-add_priority_fields_to_tasks.py
new file mode 100644
index 0000000..5b12b61
--- /dev/null
+++ b/backend/alembic/versions/2026_04_24_0000-add_priority_fields_to_tasks.py
@@ -0,0 +1,35 @@
+"""Add important and urgent priority fields to tasks
+
+Revision ID: a1b2c3d4e5f6
+Revises: 2fd06db2a9d8
+Create Date: 2026-04-24 00:00:00.000000
+"""
+from collections.abc import Sequence
+from typing import Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = 'a1b2c3d4e5f6'
+down_revision: Union[str, Sequence[str], None] = '2fd06db2a9d8'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column('tasks', sa.Column('important', sa.Boolean(), nullable=False, server_default='false'))
+    op.add_column('tasks', sa.Column('urgent', sa.Boolean(), nullable=False, server_default='false'))
+    # Drop server_defaults after backfill so model and DB stay in sync
+    op.alter_column('tasks', 'important', server_default=None)
+    op.alter_column('tasks', 'urgent', server_default=None)
+    op.create_index(
+        'ix_tasks_priority_lookup',
+        'tasks',
+        ['user_id', 'status', 'bucket', 'important', 'urgent'],
+    )
+
+
+def downgrade() -> None:
+    op.drop_index('ix_tasks_priority_lookup', table_name='tasks')
+    op.drop_column('tasks', 'urgent')
+    op.drop_column('tasks', 'important')
diff --git a/backend/app/api/state.py b/backend/app/api/state.py
new file mode 100644
index 0000000..51f8479
--- /dev/null
+++ b/backend/app/api/state.py
@@ -0,0 +1,50 @@
+import uuid
+
+from fastapi import APIRouter, Depends
+from pydantic import BaseModel
+from sqlalchemy import func
+from sqlmodel import Session, select
+
+from app.core.deps import get_db
+from app.core.security import get_current_user_id
+from app.models.enums import TaskStatus
+from app.models.task import Task
+
+router = APIRouter(prefix="/state", tags=["state"])
+
+
+class PriorityCounts(BaseModel):
+    q1_count: int
+    q2_count: int
+    q3_count: int
+    q4_count: int
+
+
+class StateResponse(BaseModel):
+    priority: PriorityCounts
+
+
+@router.get("", response_model=StateResponse)
+def get_state(
+    db: Session = Depends(get_db),
+    user_id: uuid.UUID = Depends(get_current_user_id),
+):
+    rows = db.exec(
+        select(Task.important, Task.urgent, func.count())
+        .where(
+            Task.user_id == user_id,
+            Task.status == TaskStatus.pending,
+            Task.parent_id.is_(None),
+        )
+        .group_by(Task.important, Task.urgent)
+    ).all()
+
+    counts: dict[tuple[bool, bool], int] = {(imp, urg): cnt for imp, urg, cnt in rows}
+    return StateResponse(
+        priority=PriorityCounts(
+            q1_count=counts.get((True, True), 0),
+            q2_count=counts.get((True, False), 0),
+            q3_count=counts.get((False, True), 0),
+            q4_count=counts.get((False, False), 0),
+        )
+    )
diff --git a/backend/app/api/tasks.py b/backend/app/api/tasks.py
index ca5b813..b3e73fd 100644
--- a/backend/app/api/tasks.py
+++ b/backend/app/api/tasks.py
@@ -9,6 +9,7 @@
 from app.models.enums import BucketType, TaskStatus
 from app.schemas.task_schemas import (
     DomainBrief,
+    PriorityUpdate,
     ReorderRequest,
     SubTaskResponse,
     TaskCreate,
@@ -41,6 +42,8 @@ def _to_response(task, mit_task_id: uuid.UUID | None = None) -> TaskResponse:
         updated_at=task.updated_at,
         completed_at=task.completed_at,
         is_mit=mit_task_id is not None and task.id == mit_task_id,
+        important=task.important,
+        urgent=task.urgent,
     )
 
 
@@ -91,6 +94,8 @@ def create_task(
         parent_id=body.parent_id,
         notes=body.notes,
         skip_triage_stamp=skip_stamp,
+        important=body.important,
+        urgent=body.urgent,
     )
     # Reload with relationships for response
     task = task_service.get_task(db, user_id, task.id)
@@ -125,6 +130,26 @@ def update_task(
         kwargs["status"] = body.status
     if "notes" in body.model_fields_set:
         kwargs["notes"] = body.notes
+    if body.important is not None:
+        kwargs["important"] = body.important
+    if body.urgent is not None:
+        kwargs["urgent"] = body.urgent
+    task = task_service.update_task(db, user_id, task_id, **kwargs)
+    return _to_response(task)
+
+
+@router.post("/{task_id}/priority", response_model=TaskResponse)
+def set_priority(
+    task_id: uuid.UUID,
+    body: PriorityUpdate,
+    db: Session = Depends(get_db),
+    user_id: uuid.UUID = Depends(get_current_user_id),
+):
+    kwargs: dict = {}
+    if body.important is not None:
+        kwargs["important"] = body.important
+    if body.urgent is not None:
+        kwargs["urgent"] = body.urgent
     task = task_service.update_task(db, user_id, task_id, **kwargs)
     return _to_response(task)
 
diff --git a/backend/app/main.py b/backend/app/main.py
index db72fe0..3c1e1a3 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -6,7 +6,7 @@
 from slowapi.errors import RateLimitExceeded
 from sqlalchemy import text
 
-from app.api import account, billing, domains, tasks, triage
+from app.api import account, billing, domains, state, tasks, triage
 from app.core.config import settings
 from app.core.deps import engine
 from app.core.errors import AppError, app_error_handler
@@ -23,6 +23,7 @@
 app.include_router(domains.router)
 app.include_router(account.router)
 app.include_router(billing.router)
+app.include_router(state.router)
 
 # Error handlers
 app.add_exception_handler(AppError, app_error_handler)
diff --git a/backend/app/models/task.py b/backend/app/models/task.py
index 31f53d0..5b4969d 100644
--- a/backend/app/models/task.py
+++ b/backend/app/models/task.py
@@ -2,7 +2,7 @@
 from datetime import date, datetime
 from typing import Optional
 
-from sqlalchemy import Column, String
+from sqlalchemy import Column, Index, String
 from sqlmodel import Field, Relationship, SQLModel
 
 from app.models.enums import BucketType, TaskStatus
@@ -10,6 +10,9 @@
 
 class Task(SQLModel, table=True):
     __tablename__ = "tasks"
+    __table_args__ = (
+        Index("ix_tasks_priority_lookup", "user_id", "status", "bucket", "important", "urgent"),
+    )
 
     id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
     user_id: uuid.UUID = Field(foreign_key="users.id", ondelete="CASCADE")
@@ -20,6 +23,8 @@ class Task(SQLModel, table=True):
     notes: str | None = Field(default=None, max_length=2000)
     position: int | None = Field(default=None)
     triaged_at: date | None = Field(default=None)
+    important: bool = Field(default=False, nullable=False)
+    urgent: bool = Field(default=False, nullable=False)
 
     # Parent task (one level of sub-tasks)
     parent_id: uuid.UUID | None = Field(
diff --git a/backend/app/schemas/task_schemas.py b/backend/app/schemas/task_schemas.py
index 936f8dd..375e1bd 100644
--- a/backend/app/schemas/task_schemas.py
+++ b/backend/app/schemas/task_schemas.py
@@ -6,6 +6,11 @@
 from app.models.enums import BucketType, TaskStatus
 
 
+class PriorityUpdate(BaseModel):
+    important: bool | None = None
+    urgent: bool | None = None
+
+
 class TaskCreate(BaseModel):
     text: str
     bucket: BucketType = BucketType.today
@@ -13,6 +18,8 @@ class TaskCreate(BaseModel):
     parent_id: uuid.UUID | None = None
     notes: str | None = None
     skip_triage_stamp: bool = False
+    important: bool = False
+    urgent: bool = False
 
 
 class TaskUpdate(BaseModel):
@@ -21,6 +28,8 @@ class TaskUpdate(BaseModel):
     domain_id: uuid.UUID | None = None
     status: TaskStatus | None = None
     notes: str | None = None
+    important: bool | None = None
+    urgent: bool | None = None
 
 
 class ReorderRequest(BaseModel):
@@ -64,6 +73,8 @@ class TaskResponse(BaseModel):
     updated_at: datetime
     completed_at: datetime | None
     is_mit: bool = False
+    important: bool = False
+    urgent: bool = False
 
     @computed_field
     @property
diff --git a/backend/app/services/task_service.py b/backend/app/services/task_service.py
index 97ef7c7..3a2c325 100644
--- a/backend/app/services/task_service.py
+++ b/backend/app/services/task_service.py
@@ -31,6 +31,8 @@ def create_task(
     parent_id: uuid.UUID | None = None,
     notes: str | None = None,
     skip_triage_stamp: bool = False,
+    important: bool = False,
+    urgent: bool = False,
 ) -> Task:
     if len(text) > 500:
         raise AppError(
@@ -101,6 +103,8 @@ def create_task(
         notes=notes,
         position=position,
         triaged_at=None if skip_triage_stamp else date.today(),
+        important=important,
+        urgent=urgent,
     )
     db.add(task)
     db.flush()
@@ -160,6 +164,8 @@ def update_task(
     domain_id: uuid.UUID | None | object = _UNSET,
     status: TaskStatus | None = None,
     notes: str | None | object = _UNSET,
+    important: bool | None = None,
+    urgent: bool | None = None,
 ) -> Task:
     task = get_task(db, user_id, task_id)
 
@@ -195,6 +201,10 @@ def update_task(
         task.bucket = bucket
     if domain_id is not _UNSET:
         task.domain_id = domain_id
+    if important is not None:
+        task.important = important
+    if urgent is not None:
+        task.urgent = urgent
     if status is not None:
         allowed = {
             TaskStatus.pending: {TaskStatus.archived},
diff --git a/backend/tests/test_tasks.py b/backend/tests/test_tasks.py
index c04558a..fedae0c 100644
--- a/backend/tests/test_tasks.py
+++ b/backend/tests/test_tasks.py
@@ -398,3 +398,130 @@ def test_notes_preserved_on_unrelated_update(self, client, test_user, test_tasks
         r = client.patch(f"/tasks/{task.id}", json={"text": "New text"})
         assert r.status_code == 200
         assert r.json()["notes"] == "Keep this"
+
+
+class TestPriorityFields:
+    def test_create_task_with_priority(self, client, test_user):
+        r = client.post("/tasks", json={"text": "Q1 task", "important": True, "urgent": True})
+        assert r.status_code == 201
+        data = r.json()
+        assert data["important"] is True
+        assert data["urgent"] is True
+
+    def test_create_task_default_priority(self, client, test_user):
+        r = client.post("/tasks", json={"text": "Plain task"})
+        assert r.status_code == 201
+        data = r.json()
+        assert data["important"] is False
+        assert data["urgent"] is False
+
+    def test_patch_important(self, client, test_user, test_tasks):
+        task = test_tasks[0]
+        r = client.patch(f"/tasks/{task.id}", json={"important": True})
+        assert r.status_code == 200
+        assert r.json()["important"] is True
+        assert r.json()["urgent"] is False  # unchanged
+
+    def test_priority_endpoint_toggle(self, client, test_user, test_tasks):
+        task = test_tasks[0]
+        r = client.post(f"/tasks/{task.id}/priority", json={"important": True})
+        assert r.status_code == 200
+        assert r.json()["important"] is True
+        assert r.json()["urgent"] is False
+
+    def test_priority_endpoint_partial_update(self, client, test_user, test_tasks):
+        """Partial payload doesn't reset the other field."""
+        task = test_tasks[0]
+        client.post(f"/tasks/{task.id}/priority", json={"important": True, "urgent": True})
+        r = client.post(f"/tasks/{task.id}/priority", json={"urgent": False})
+        assert r.status_code == 200
+        assert r.json()["important"] is True   # untouched
+        assert r.json()["urgent"] is False
+
+    def test_priority_endpoint_idempotent(self, client, test_user, test_tasks):
+        task = test_tasks[0]
+        client.post(f"/tasks/{task.id}/priority", json={"important": True})
+        r = client.post(f"/tasks/{task.id}/priority", json={"important": True})
+        assert r.status_code == 200
+        assert r.json()["important"] is True
+
+
+class TestStateEndpoint:
+    def test_state_priority_counts(self, client, test_user, db):
+        # Create tasks in each quadrant
+        from app.models.task import Task as TaskModel
+        from app.models.enums import BucketType, TaskStatus
+        import uuid
+        user_id = uuid.UUID("00000000-0000-0000-0000-000000000099")
+
+        for important, urgent in [(True, True), (True, False), (False, True), (False, False)]:
+            t = TaskModel(user_id=user_id, text="t", bucket=BucketType.today,
+                          status=TaskStatus.pending, important=important, urgent=urgent)
+            db.add(t)
+        db.flush()
+
+        r = client.get("/state")
+        assert r.status_code == 200
+        p = r.json()["priority"]
+        assert p["q1_count"] == 1
+        assert p["q2_count"] == 1
+        assert p["q3_count"] == 1
+        assert p["q4_count"] == 1
+
+    def test_state_excludes_completed(self, client, test_user, db):
+        from app.models.task import Task as TaskModel
+        from app.models.enums import BucketType, TaskStatus
+        import uuid
+        user_id = uuid.UUID("00000000-0000-0000-0000-000000000099")
+
+        t = TaskModel(user_id=user_id, text="done", bucket=BucketType.today,
+                      status=TaskStatus.complete, important=True, urgent=True)
+        db.add(t)
+        db.flush()
+
+        r = client.get("/state")
+        assert r.status_code == 200
+        assert r.json()["priority"]["q1_count"] == 0
+
+    def test_state_excludes_subtasks(self, client, test_user, db):
+        """Sub-tasks with priority flags must not appear in quadrant counts."""
+        from app.models.task import Task as TaskModel
+        from app.models.enums import BucketType, TaskStatus
+        import uuid
+        user_id = uuid.UUID("00000000-0000-0000-0000-000000000099")
+
+        parent = TaskModel(user_id=user_id, text="parent", bucket=BucketType.today,
+                           status=TaskStatus.pending)
+        db.add(parent)
+        db.flush()
+
+        child = TaskModel(user_id=user_id, text="child", bucket=BucketType.today,
+                          status=TaskStatus.pending, parent_id=parent.id,
+                          important=True, urgent=True)
+        db.add(child)
+        db.flush()
+
+        r = client.get("/state")
+        assert r.status_code == 200
+        # Sub-task must not appear in q1_count
+        assert r.json()["priority"]["q1_count"] == 0
+
+    def test_state_scoped_to_current_user(self, client, test_user, db):
+        """Tasks from another user must not appear in counts."""
+        from app.models.task import Task as TaskModel
+        from app.models.user import User
+        from app.models.enums import BucketType, TaskStatus, AuthProvider
+        import uuid
+
+        other_id = uuid.uuid4()
+        other = User(id=other_id, email="other@tend.app", auth_provider=AuthProvider.email)
+        db.add(other)
+
+        t = TaskModel(user_id=other_id, text="other", bucket=BucketType.today,
+                      status=TaskStatus.pending, important=True, urgent=True)
+        db.add(t)
+        db.flush()
+
+        r = client.get("/state")
+        assert r.status_code == 200
+        assert r.json()["priority"]["q1_count"] == 0