feat: state rm (#2880)

* feat: state rm * review feedback * fix conflict for pegomock generation code * adopt state command into allow-commands * fix conflicts * fix: state rm works on workspace * notify import/state rm discard plan file * fix lint * use repeat instead warning for re-plan * perl -pi -e 's!\* 🔁 plan file was discarded. to!🚮 A plan file was discarded. Re-plan would be required before applying.\n\n\* 🔁 To!g' server/**/* * follow main branch
2026-07-28 21:38:25 +00:00 · 2023-01-19 13:02:18 +09:00
parent 9e3d8879ec
commit cb0aadf571
81 changed files with 1793 additions and 217 deletions
--- a/go.mod
+++ b/go.mod
@@ -43,6 +43,7 @@ require (
 	github.com/xanzy/go-gitlab v0.78.0
 	go.etcd.io/bbolt v1.3.6
 	go.uber.org/zap v1.24.0
+	golang.org/x/exp v0.0.0-20230113152452-c42ee1cf562e
 	golang.org/x/term v0.4.0
 	golang.org/x/text v0.6.0
 	gopkg.in/yaml.v2 v2.4.0
--- a/go.sum
+++ b/go.sum
@@ -508,6 +508,8 @@ golang.org/x/exp v0.0.0-20191227195350-da58074b4299/go.mod h1:2RIsYlXP63K8oxa1u0
 golang.org/x/exp v0.0.0-20200119233911-0405dc783f0a/go.mod h1:2RIsYlXP63K8oxa1u096TMicItID8zy7Y6sNkU49FU4=
 golang.org/x/exp v0.0.0-20200207192155-f17229e696bd/go.mod h1:J/WKrq2StrnmMY6+EHIKF9dgMWnmCNThgcyBT1FY9mM=
 golang.org/x/exp v0.0.0-20200224162631-6cc2880d07d6/go.mod h1:3jZMyOhIsHpP37uCMkUooju7aAi5cS1Q23tOzKc+0MU=
+golang.org/x/exp v0.0.0-20230113152452-c42ee1cf562e h1:uGuXqQsI2BAE8xNqSqNxhTdDdhlvpBvWFw/KBwtCtjI=
+golang.org/x/exp v0.0.0-20230113152452-c42ee1cf562e/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/image v0.0.0-20190227222117-0694c2d4d067/go.mod h1:kZ7UVZpmo3dzQBMxlp+ypCbDeSB+sBbTgSJuh5dn5js=
 golang.org/x/image v0.0.0-20190802002840-cff245a6509b/go.mod h1:FeLwcggjj3mMvU+oOTbSwawSJRM1uh48EjtB4UJZlP0=
 golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE=
--- a/runatlantis.io/docs/custom-workflows.md
+++ b/runatlantis.io/docs/custom-workflows.md
@@ -56,6 +56,11 @@ workflows:
        - init
        - import:
            extra_args: ["-var-file", "production.tfvars"]
+    state_rm:
+      steps:
+        - init
+        - state_rm:
+            extra_args: ["-lock=false"]
 ```
 Then in your repo-level `atlantis.yaml` file, you would reference the workflows:
 ```yaml
@@ -374,13 +379,15 @@ projects:
 plan:
 apply:
 import:
+state_rm:
 ```

-| Key    | Type            | Default                 | Required | Description                      |
-|--------|-----------------|-------------------------|----------|----------------------------------|
-| plan   | [Stage](#stage) | `steps: [init, plan]`   | no       | How to plan for this project.    |
-| apply  | [Stage](#stage) | `steps: [apply]`        | no       | How to apply for this project.   |
-| import | [Stage](#stage) | `steps: [init, import]` | no       | How to import for this project.  |
+| Key      | Type            | Default                   | Required | Description                           |
+|----------|-----------------|---------------------------|----------|---------------------------------------|
+| plan     | [Stage](#stage) | `steps: [init, plan]`     | no       | How to plan for this project.         |
+| apply    | [Stage](#stage) | `steps: [apply]`          | no       | How to apply for this project.        |
+| import   | [Stage](#stage) | `steps: [init, import]`   | no       | How to import for this project.       |
+| state_rm | [Stage](#stage) | `steps: [init, state_rm]` | no       | How to run state rm for this project. |

 ### Stage
 ```yaml
@@ -403,10 +410,11 @@ Steps can be a single string for a built-in command.
 - plan
 - apply
 - import
+- state_rm
 ```
-| Key                    | Type   | Default | Required | Description                                                                                                      |
-|------------------------|--------|---------|----------|------------------------------------------------------------------------------------------------------------------|
-| init/plan/apply/import | string | none    | no       | Use a built-in command without additional configuration. Only `init`, `plan`, `apply` and `import` are supported |
+| Key                             | Type   | Default | Required | Description                                                                                                                  |
+|---------------------------------|--------|---------|----------|------------------------------------------------------------------------------------------------------------------------------|
+| init/plan/apply/import/state_rm | string | none    | no       | Use a built-in command without additional configuration. Only `init`, `plan`, `apply`, `import` and `state_rm` are supported |

 #### Built-In Command With Extra Args
 A map from string to `extra_args` for a built-in command with extra arguments.
@@ -419,10 +427,12 @@ A map from string to `extra_args` for a built-in command with extra arguments.
    extra_args: [arg1, arg2]
 - import:
    extra_args: [arg1, arg2]
+- state_rm:
+    extra_args: [arg1, arg2]
 ```
-| Key                    | Type                               | Default | Required | Description                                                                                                                                                   |
-|------------------------|------------------------------------|---------|----------|---------------------------------------------------------------------------------------------------------------------------------------------------------------|
-| init/plan/apply/import | map[`extra_args` -> array[string]] | none    | no       | Use a built-in command and append `extra_args`. Only `init`, `plan`, `apply` and `import` are supported as keys and only `extra_args` is supported as a value |
+| Key                             | Type                               | Default | Required | Description                                                                                                                                                               |
+|---------------------------------|------------------------------------|---------|----------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| init/plan/apply/import/state_rm | map[`extra_args` -> array[string]] | none    | no       | Use a built-in command and append `extra_args`. Only `init`, `plan`, `apply`, `import` and `state_rm` are supported as keys and only `extra_args` is supported as a value |

 #### Custom `run` Command
 Or a custom command
--- a/runatlantis.io/docs/server-configuration.md
+++ b/runatlantis.io/docs/server-configuration.md
@@ -57,7 +57,7 @@ Values are chosen in this order:

  Notes:
  * Accepts a comma separated list, ex. `command1,command2`.
-  * `version`, `plan`, `apply`, `unlock`, `approve_policies`, `import` and `all` are available.
+  * `version`, `plan`, `apply`, `unlock`, `approve_policies`, `import`, `state` and `all` are available.
  * `all` is a special keyword that allows all commands. If pass `all` then all other commands will be ignored.

 ### `--allow-draft-prs`
--- a/runatlantis.io/docs/using-atlantis.md
+++ b/runatlantis.io/docs/using-atlantis.md
@@ -169,6 +169,51 @@ atlantis import -d dir 'aws_instance.example["foo"]' i-1234567890abcdef0 -- -var
 ```
 If a flag is needed to be always appended, see [Custom Workflow Use Cases](custom-workflows.html#adding-extra-arguments-to-terraform-commands).

+---
+## atlantis state rm
+```bash
+atlantis state [options] rm ADDRESS... -- [terraform state rm flags]
+```
+### Explanation
+Runs `terraform state rm` that matches the directory/project/workspace.
+This command discards the terraform plan result. After run state rm and before an apply, another `atlantis plan` must be run again.
+
+To allow the `state` command requires [--allow-commands](/docs/server-configuration.html#allow-commands) configuration.
+
+### Examples
+```bash
+# Runs state rm
+atlantis state rm ADDRESS1 ADDRESS2
+
+# Runs state rm in the root directory of the repo with workspace `default`
+atlantis state -d . rm ADDRESS
+
+# Runs state rm in the `project1` directory of the repo with workspace `default`
+atlantis state -d project1 rm ADDRESS
+
+# Runs state rm in the root directory of the repo with workspace `staging`
+atlantis state -w staging rm ADDRESS
+```
+
+::: tip
+* If run state rm to for_each resources, it requires a single quoted address.
+  * ex. `atlantis state rm 'aws_instance.example["foo"]'`
+:::
+
+### Options
+* `-d directory` Run state rm a resource for this directory, relative to root of repo. Use `.` for root.
+* `-p project` Run state rm a resource for this project. Refers to the name of the project configured in the repo's [`atlantis.yaml`](repo-level-atlantis-yaml.html) repo configuration file. This cannot be used at the same time as `-d` or `-w`.
+* `-w workspace` Run state rm a resource for a specific [Terraform workspace](https://developer.hashicorp.com/terraform/language/state/workspaces). Ignore this if Terraform workspaces are unused.
+
+### Additional Terraform flags
+
+If `terraform state rm` requires additional arguments, like `-lock=false'`
+append them to the end of the comment after `--`, e.g.
+```
+atlantis state -d dir rm 'aws_instance.example["foo"]' -- -lock=false
+```
+If a flag is needed to be always appended, see [Custom Workflow Use Cases](custom-workflows.html#adding-extra-arguments-to-terraform-commands).
+
 ---
 ## atlantis unlock
 ```bash
--- a/server/controllers/events/events_controller_e2e_test.go
+++ b/server/controllers/events/events_controller_e2e_test.go
@@ -497,6 +497,71 @@ func TestGitHubWorkflow(t *testing.T) {
 				{"exp-output-merge.txt"},
 			},
 		},
+		{
+			Description:   "state rm single project",
+			RepoDir:       "state-rm-single-project",
+			ModifiedFiles: []string{"main.tf"},
+			ExpAutoplan:   true,
+			Comments: []string{
+				"atlantis import random_id.simple AA",
+				"atlantis import 'random_id.for_each[\"overridden\"]' BB -- -var var=overridden",
+				"atlantis import random_id.count[0] BB",
+				"atlantis plan -- -var var=overridden",
+				"atlantis state rm 'random_id.for_each[\"overridden\"]' -- -lock=false",
+				"atlantis state rm random_id.count[0] random_id.simple",
+				"atlantis plan -- -var var=overridden",
+			},
+			ExpReplies: [][]string{
+				{"exp-output-autoplan.txt"},
+				{"exp-output-import-simple.txt"},
+				{"exp-output-import-foreach.txt"},
+				{"exp-output-import-count.txt"},
+				{"exp-output-plan.txt"},
+				{"exp-output-state-rm-foreach.txt"},
+				{"exp-output-state-rm-multiple.txt"},
+				{"exp-output-plan-again.txt"},
+				{"exp-output-merged.txt"},
+			},
+		},
+		{
+			Description: "state rm workspace",
+			RepoDir:     "state-rm-workspace",
+			Comments: []string{
+				"atlantis import -p dir1-ops 'random_id.dummy1[0]' AA",
+				"atlantis plan -p dir1-ops",
+				"atlantis state rm -p dir1-ops 'random_id.dummy1[0]'",
+				"atlantis plan -p dir1-ops",
+			},
+			ExpReplies: [][]string{
+				{"exp-output-import-dummy1.txt"},
+				{"exp-output-plan.txt"},
+				{"exp-output-state-rm-dummy1.txt"},
+				{"exp-output-plan-again.txt"},
+				{"exp-output-merge.txt"},
+			},
+		},
+		{
+			Description:   "state rm multiple project",
+			RepoDir:       "state-rm-multiple-project",
+			ModifiedFiles: []string{"dir1/main.tf", "dir2/main.tf"},
+			ExpAutoplan:   true,
+			Comments: []string{
+				"atlantis import -d dir1 random_id.dummy AA",
+				"atlantis import -d dir2 random_id.dummy BB",
+				"atlantis plan",
+				"atlantis state rm random_id.dummy",
+				"atlantis plan",
+			},
+			ExpReplies: [][]string{
+				{"exp-output-autoplan.txt"},
+				{"exp-output-import-dummy1.txt"},
+				{"exp-output-import-dummy2.txt"},
+				{"exp-output-plan.txt"},
+				{"exp-output-state-rm-multiple-projects.txt"},
+				{"exp-output-plan-again.txt"},
+				{"exp-output-merged.txt"},
+			},
+		},
 	}
 	for _, c := range cases {
 		t.Run(c.Description, func(t *testing.T) {
@@ -1152,7 +1217,8 @@ func setupE2E(t *testing.T, repoDir string, opt setupOption) (events_controllers
 		ApplyStepRunner: &runtime.ApplyStepRunner{
 			TerraformExecutor: terraformClient,
 		},
-		ImportStepRunner: runtime.NewImportStepRunner(terraformClient, defaultTFVersion),
+		ImportStepRunner:  runtime.NewImportStepRunner(terraformClient, defaultTFVersion),
+		StateRmStepRunner: runtime.NewStateRmStepRunner(terraformClient, defaultTFVersion),
 		RunStepRunner: &runtime.RunStepRunner{
 			TerraformExecutor:       terraformClient,
 			DefaultTFVersion:        defaultTFVersion,
@@ -1262,6 +1328,12 @@ func setupE2E(t *testing.T, repoDir string, opt setupOption) (events_controllers
 		projectCommandRunner,
 	)

+	stateCommandRunner := events.NewStateCommandRunner(
+		pullUpdater,
+		projectCommandBuilder,
+		projectCommandRunner,
+	)
+
 	commentCommandRunnerByCmd := map[command.Name]events.CommentCommandRunner{
 		command.Plan:            planCommandRunner,
 		command.Apply:           applyCommandRunner,
@@ -1269,6 +1341,7 @@ func setupE2E(t *testing.T, repoDir string, opt setupOption) (events_controllers
 		command.Unlock:          unlockCommandRunner,
 		command.Version:         versionCommandRunner,
 		command.Import:          importCommandRunner,
+		command.State:           stateCommandRunner,
 	}

 	commandRunner := &events.DefaultCommandRunner{
--- a/server/controllers/events/testdata/test-repos/import-multiple-project/exp-output-import-dummy1.txt
+++ b/server/controllers/events/testdata/test-repos/import-multiple-project/exp-output-import-dummy1.txt
@@ -12,5 +12,7 @@ The resources that were imported are shown above. These resources are now in
 your Terraform state and will henceforth be managed by Terraform.
 ```

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * `atlantis plan -d dir1`
--- a/server/controllers/events/testdata/test-repos/import-single-project-var/exp-output-import-count.txt
+++ b/server/controllers/events/testdata/test-repos/import-single-project-var/exp-output-import-count.txt
@@ -12,5 +12,7 @@ The resources that were imported are shown above. These resources are now in
 your Terraform state and will henceforth be managed by Terraform.
 ```

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/import-single-project-var/exp-output-import-foreach.txt
+++ b/server/controllers/events/testdata/test-repos/import-single-project-var/exp-output-import-foreach.txt
@@ -12,5 +12,7 @@ The resources that were imported are shown above. These resources are now in
 your Terraform state and will henceforth be managed by Terraform.
 ```

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/import-single-project/exp-output-import-dummy1.txt
+++ b/server/controllers/events/testdata/test-repos/import-single-project/exp-output-import-dummy1.txt
@@ -12,5 +12,7 @@ The resources that were imported are shown above. These resources are now in
 your Terraform state and will henceforth be managed by Terraform.
 ```

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/import-single-project/exp-output-import-dummy2.txt
+++ b/server/controllers/events/testdata/test-repos/import-single-project/exp-output-import-dummy2.txt
@@ -12,5 +12,7 @@ The resources that were imported are shown above. These resources are now in
 your Terraform state and will henceforth be managed by Terraform.
 ```

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/import-workspace/exp-output-import-dir1-ops-dummy1.txt
+++ b/server/controllers/events/testdata/test-repos/import-workspace/exp-output-import-dir1-ops-dummy1.txt
@@ -12,5 +12,7 @@ The resources that were imported are shown above. These resources are now in
 your Terraform state and will henceforth be managed by Terraform.
 ```

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * `atlantis plan -p dir1-ops`
--- a/server/controllers/events/testdata/test-repos/import-workspace/exp-output-import-dir1-ops-dummy2.txt
+++ b/server/controllers/events/testdata/test-repos/import-workspace/exp-output-import-dir1-ops-dummy2.txt
@@ -12,5 +12,7 @@ The resources that were imported are shown above. These resources are now in
 your Terraform state and will henceforth be managed by Terraform.
 ```

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * `atlantis plan -p dir1-ops`
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/atlantis.yaml
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/atlantis.yaml
@@ -0,0 +1,4 @@
+version: 3
+projects:
+- dir: dir1
+- dir: dir2
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/dir1/main.tf
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/dir1/main.tf
@@ -0,0 +1,4 @@
+resource "random_id" "dummy" {
+  keepers     = {}
+  byte_length = 1
+}
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/dir2/main.tf
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/dir2/main.tf
@@ -0,0 +1,4 @@
+resource "random_id" "dummy" {
+  keepers     = {}
+  byte_length = 1
+}
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-autoplan.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-autoplan.txt
@@ -0,0 +1,75 @@
+Ran Plan for 2 projects:
+
+1. dir: `dir1` workspace: `default`
+1. dir: `dir2` workspace: `default`
+
+### 1. dir: `dir1` workspace: `default`
+<details><summary>Show Output</summary>
+
+```diff
+Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+ create
+
+Terraform will perform the following actions:
+
+  # random_id.dummy will be created
+ resource "random_id" "dummy" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+Plan: 1 to add, 0 to change, 0 to destroy.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d dir1`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d dir1`
+</details>
+Plan: 1 to add, 0 to change, 0 to destroy.
+
+---
+### 2. dir: `dir2` workspace: `default`
+<details><summary>Show Output</summary>
+
+```diff
+Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+ create
+
+Terraform will perform the following actions:
+
+  # random_id.dummy will be created
+ resource "random_id" "dummy" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+Plan: 1 to add, 0 to change, 0 to destroy.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d dir2`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d dir2`
+</details>
+Plan: 1 to add, 0 to change, 0 to destroy.
+
+---
+* :fast_forward: To **apply** all unapplied plans from this pull request, comment:
+    * `atlantis apply`
+* :put_litter_in_its_place: To delete all plans and locks for the PR, comment:
+    * `atlantis unlock`
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-import-dummy1.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-import-dummy1.txt
@@ -0,0 +1,18 @@
+Ran Import for dir: `dir1` workspace: `default`
+
+```diff
+random_id.dummy: Importing from ID "AA"...
+random_id.dummy: Import prepared!
+  Prepared random_id for import
+random_id.dummy: Refreshing state... [id=AA]
+
+Import successful!
+
+The resources that were imported are shown above. These resources are now in
+your Terraform state and will henceforth be managed by Terraform.
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d dir1`
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-import-dummy2.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-import-dummy2.txt
@@ -0,0 +1,18 @@
+Ran Import for dir: `dir2` workspace: `default`
+
+```diff
+random_id.dummy: Importing from ID "BB"...
+random_id.dummy: Import prepared!
+  Prepared random_id for import
+random_id.dummy: Refreshing state... [id=BB]
+
+Import successful!
+
+The resources that were imported are shown above. These resources are now in
+your Terraform state and will henceforth be managed by Terraform.
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d dir2`
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-merged.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-merged.txt
@@ -0,0 +1,4 @@
+Locks and plans deleted for the projects and workspaces modified in this pull request:
+
+- dir: `dir1` workspace: `default`
+- dir: `dir2` workspace: `default`
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-plan-again.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-plan-again.txt
@@ -0,0 +1,75 @@
+Ran Plan for 2 projects:
+
+1. dir: `dir1` workspace: `default`
+1. dir: `dir2` workspace: `default`
+
+### 1. dir: `dir1` workspace: `default`
+<details><summary>Show Output</summary>
+
+```diff
+Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+ create
+
+Terraform will perform the following actions:
+
+  # random_id.dummy will be created
+ resource "random_id" "dummy" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+Plan: 1 to add, 0 to change, 0 to destroy.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d dir1`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d dir1`
+</details>
+Plan: 1 to add, 0 to change, 0 to destroy.
+
+---
+### 2. dir: `dir2` workspace: `default`
+<details><summary>Show Output</summary>
+
+```diff
+Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+ create
+
+Terraform will perform the following actions:
+
+  # random_id.dummy will be created
+ resource "random_id" "dummy" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+Plan: 1 to add, 0 to change, 0 to destroy.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d dir2`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d dir2`
+</details>
+Plan: 1 to add, 0 to change, 0 to destroy.
+
+---
+* :fast_forward: To **apply** all unapplied plans from this pull request, comment:
+    * `atlantis apply`
+* :put_litter_in_its_place: To delete all plans and locks for the PR, comment:
+    * `atlantis unlock`
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-plan.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-plan.txt
@@ -0,0 +1,43 @@
+Ran Plan for 2 projects:
+
+1. dir: `dir1` workspace: `default`
+1. dir: `dir2` workspace: `default`
+
+### 1. dir: `dir1` workspace: `default`
+```diff
+random_id.dummy: Refreshing state... [id=AA]
+
+No changes. Your infrastructure matches the configuration.
+
+Terraform has compared your real infrastructure against your configuration
+and found no differences, so no changes are needed.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d dir1`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d dir1`
+
+---
+### 2. dir: `dir2` workspace: `default`
+```diff
+random_id.dummy: Refreshing state... [id=BB]
+
+No changes. Your infrastructure matches the configuration.
+
+Terraform has compared your real infrastructure against your configuration
+and found no differences, so no changes are needed.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d dir2`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d dir2`
+
+---
+* :fast_forward: To **apply** all unapplied plans from this pull request, comment:
+    * `atlantis apply`
+* :put_litter_in_its_place: To delete all plans and locks for the PR, comment:
+    * `atlantis unlock`
--- a/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-state-rm-multiple-projects.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-multiple-project/exp-output-state-rm-multiple-projects.txt
@@ -0,0 +1,29 @@
+Ran State for 2 projects:
+
+1. dir: `dir1` workspace: `default`
+1. dir: `dir2` workspace: `default`
+
+### 1. dir: `dir1` workspace: `default`
+```diff
+Removed random_id.dummy
+Successfully removed 1 resource instance(s).
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d dir1`
+
+---
+### 2. dir: `dir2` workspace: `default`
+```diff
+Removed random_id.dummy
+Successfully removed 1 resource instance(s).
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d dir2`
+
+---
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-autoplan.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-autoplan.txt
@@ -0,0 +1,60 @@
+Ran Plan for dir: `.` workspace: `default`
+
+<details><summary>Show Output</summary>
+
+```diff
+Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+ create
+
+Terraform will perform the following actions:
+
+  # random_id.count[0] will be created
+ resource "random_id" "count" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+  # random_id.for_each["default"] will be created
+ resource "random_id" "for_each" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+  # random_id.simple will be created
+ resource "random_id" "simple" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+Plan: 3 to add, 0 to change, 0 to destroy.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d .`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d .`
+</details>
+Plan: 3 to add, 0 to change, 0 to destroy.
+
+---
+* :fast_forward: To **apply** all unapplied plans from this pull request, comment:
+    * `atlantis apply`
+* :put_litter_in_its_place: To delete all plans and locks for the PR, comment:
+    * `atlantis unlock`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-import-count.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-import-count.txt
@@ -0,0 +1,18 @@
+Ran Import for dir: `.` workspace: `default`
+
+```diff
+random_id.count[0]: Importing from ID "BB"...
+random_id.count[0]: Import prepared!
+  Prepared random_id for import
+random_id.count[0]: Refreshing state... [id=BB]
+
+Import successful!
+
+The resources that were imported are shown above. These resources are now in
+your Terraform state and will henceforth be managed by Terraform.
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-import-foreach.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-import-foreach.txt
@@ -0,0 +1,18 @@
+Ran Import for dir: `.` workspace: `default`
+
+```diff
+random_id.for_each["overridden"]: Importing from ID "BB"...
+random_id.for_each["overridden"]: Import prepared!
+  Prepared random_id for import
+random_id.for_each["overridden"]: Refreshing state... [id=BB]
+
+Import successful!
+
+The resources that were imported are shown above. These resources are now in
+your Terraform state and will henceforth be managed by Terraform.
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-import-simple.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-import-simple.txt
@@ -0,0 +1,18 @@
+Ran Import for dir: `.` workspace: `default`
+
+```diff
+random_id.simple: Importing from ID "AA"...
+random_id.simple: Import prepared!
+  Prepared random_id for import
+random_id.simple: Refreshing state... [id=AA]
+
+Import successful!
+
+The resources that were imported are shown above. These resources are now in
+your Terraform state and will henceforth be managed by Terraform.
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-merged.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-merged.txt
@@ -0,0 +1,3 @@
+Locks and plans deleted for the projects and workspaces modified in this pull request:
+
+- dir: `.` workspace: `default`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-plan-again.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-plan-again.txt
@@ -0,0 +1,60 @@
+Ran Plan for dir: `.` workspace: `default`
+
+<details><summary>Show Output</summary>
+
+```diff
+Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+ create
+
+Terraform will perform the following actions:
+
+  # random_id.count[0] will be created
+ resource "random_id" "count" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+  # random_id.for_each["overridden"] will be created
+ resource "random_id" "for_each" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+  # random_id.simple will be created
+ resource "random_id" "simple" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+Plan: 3 to add, 0 to change, 0 to destroy.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d .`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d . -- -var var=overridden`
+</details>
+Plan: 3 to add, 0 to change, 0 to destroy.
+
+---
+* :fast_forward: To **apply** all unapplied plans from this pull request, comment:
+    * `atlantis apply`
+* :put_litter_in_its_place: To delete all plans and locks for the PR, comment:
+    * `atlantis unlock`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-plan.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-plan.txt
@@ -0,0 +1,20 @@
+Ran Plan for dir: `.` workspace: `default`
+
+```diff
+No changes. Your infrastructure matches the configuration.
+
+Terraform has compared your real infrastructure against your configuration
+and found no differences, so no changes are needed.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -d .`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -d . -- -var var=overridden`
+
+---
+* :fast_forward: To **apply** all unapplied plans from this pull request, comment:
+    * `atlantis apply`
+* :put_litter_in_its_place: To delete all plans and locks for the PR, comment:
+    * `atlantis unlock`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-state-rm-foreach.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-state-rm-foreach.txt
@@ -0,0 +1,11 @@
+Ran State `rm` for dir: `.` workspace: `default`
+
+```diff
+Removed random_id.for_each["overridden"]
+Successfully removed 1 resource instance(s).
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-state-rm-multiple.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/exp-output-state-rm-multiple.txt
@@ -0,0 +1,12 @@
+Ran State `rm` for dir: `.` workspace: `default`
+
+```diff
+Removed random_id.count[0]
+Removed random_id.simple
+Successfully removed 2 resource instance(s).
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -d .`
--- a/server/controllers/events/testdata/test-repos/state-rm-single-project/main.tf
+++ b/server/controllers/events/testdata/test-repos/state-rm-single-project/main.tf
@@ -0,0 +1,20 @@
+resource "random_id" "simple" {
+  keepers     = {}
+  byte_length = 1
+}
+
+resource "random_id" "for_each" {
+  for_each    = toset([var.var])
+  keepers     = {}
+  byte_length = 1
+}
+
+resource "random_id" "count" {
+  count       = 1
+  keepers     = {}
+  byte_length = 1
+}
+
+variable "var" {
+  default = "default"
+}
--- a/server/controllers/events/testdata/test-repos/state-rm-workspace/atlantis.yaml
+++ b/server/controllers/events/testdata/test-repos/state-rm-workspace/atlantis.yaml
@@ -0,0 +1,5 @@
+version: 3
+projects:
+- name: dir1-ops
+  dir: dir1
+  workspace: ops
--- a/server/controllers/events/testdata/test-repos/state-rm-workspace/dir1/main.tf
+++ b/server/controllers/events/testdata/test-repos/state-rm-workspace/dir1/main.tf
@@ -0,0 +1,10 @@
+resource "random_id" "dummy1" {
+  count = terraform.workspace == "ops" ? 1 : 0
+
+  keepers     = {}
+  byte_length = 1
+}
+
+output "workspace" {
+  value = terraform.workspace
+}
--- a/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-import-dummy1.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-import-dummy1.txt
@@ -0,0 +1,18 @@
+Ran Import for project: `dir1-ops` dir: `dir1` workspace: `ops`
+
+```diff
+random_id.dummy1[0]: Importing from ID "AA"...
+random_id.dummy1[0]: Import prepared!
+  Prepared random_id for import
+random_id.dummy1[0]: Refreshing state... [id=AA]
+
+Import successful!
+
+The resources that were imported are shown above. These resources are now in
+your Terraform state and will henceforth be managed by Terraform.
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -p dir1-ops`
--- a/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-merge.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-merge.txt
@@ -0,0 +1,3 @@
+Locks and plans deleted for the projects and workspaces modified in this pull request:
+
+- dir: `dir1` workspace: `ops`
--- a/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-plan-again.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-plan-again.txt
@@ -0,0 +1,38 @@
+Ran Plan for project: `dir1-ops` dir: `dir1` workspace: `ops`
+
+<details><summary>Show Output</summary>
+
+```diff
+Terraform used the selected providers to generate the following execution
+plan. Resource actions are indicated with the following symbols:
+ create
+
+Terraform will perform the following actions:
+
+  # random_id.dummy1[0] will be created
+ resource "random_id" "dummy1" {
+      + b64_std     = (known after apply)
+      + b64_url     = (known after apply)
+      + byte_length = 1
+      + dec         = (known after apply)
+      + hex         = (known after apply)
+      + id          = (known after apply)
+      + keepers     = {}
+    }
+
+Plan: 1 to add, 0 to change, 0 to destroy.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -p dir1-ops`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -p dir1-ops`
+</details>
+Plan: 1 to add, 0 to change, 0 to destroy.
+
+---
+* :fast_forward: To **apply** all unapplied plans from this pull request, comment:
+    * `atlantis apply`
+* :put_litter_in_its_place: To delete all plans and locks for the PR, comment:
+    * `atlantis unlock`
--- a/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-plan.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-plan.txt
@@ -0,0 +1,22 @@
+Ran Plan for project: `dir1-ops` dir: `dir1` workspace: `ops`
+
+```diff
+random_id.dummy1[0]: Refreshing state... [id=AA]
+
+No changes. Your infrastructure matches the configuration.
+
+Terraform has compared your real infrastructure against your configuration
+and found no differences, so no changes are needed.
+```
+
+* :arrow_forward: To **apply** this plan, comment:
+    * `atlantis apply -p dir1-ops`
+* :put_litter_in_its_place: To **delete** this plan click [here](lock-url)
+* :repeat: To **plan** this project again, comment:
+    * `atlantis plan -p dir1-ops`
+
+---
+* :fast_forward: To **apply** all unapplied plans from this pull request, comment:
+    * `atlantis apply`
+* :put_litter_in_its_place: To delete all plans and locks for the PR, comment:
+    * `atlantis unlock`
--- a/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-state-rm-dummy1.txt
+++ b/server/controllers/events/testdata/test-repos/state-rm-workspace/exp-output-state-rm-dummy1.txt
@@ -0,0 +1,11 @@
+Ran State `rm` for project: `dir1-ops` dir: `dir1` workspace: `ops`
+
+```diff
+Removed random_id.dummy1[0]
+Successfully removed 1 resource instance(s).
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `atlantis plan -p dir1-ops`
--- a/server/core/config/parser_validator_test.go
+++ b/server/core/config/parser_validator_test.go
@@ -175,7 +175,8 @@ workflows:
 								},
 							},
 						},
-						Import: valid.DefaultImportStage,
+						Import:  valid.DefaultImportStage,
+						StateRm: valid.DefaultStateRmStage,
 					},
 				},
 			},
@@ -755,6 +756,9 @@ workflows:
    import:
      steps:
      - import
+    state_rm:
+      steps:
+      - state_rm
 `,
 			exp: valid.RepoCfg{
 				Version: 3,
@@ -808,6 +812,13 @@ workflows:
 								},
 							},
 						},
+						StateRm: valid.Stage{
+							Steps: []valid.Step{
+								{
+									StepName: "state_rm",
+								},
+							},
+						},
 					},
 				},
 			},
@@ -843,6 +854,10 @@ workflows:
      steps:
      - import:
          extra_args: ["a", "b"]
+    state_rm:
+      steps:
+      - state_rm:
+          extra_args: ["a", "b"]
 `,
 			exp: valid.RepoCfg{
 				Version: 3,
@@ -899,6 +914,14 @@ workflows:
 								},
 							},
 						},
+						StateRm: valid.Stage{
+							Steps: []valid.Step{
+								{
+									StepName:  "state_rm",
+									ExtraArgs: []string{"a", "b"},
+								},
+							},
+						},
 					},
 				},
 			},
@@ -923,6 +946,9 @@ workflows:
    import:
      steps:
      - run: echo apply "arg 3"
+    state_rm:
+      steps:
+      - run: echo apply "arg 4"
 `,
 			exp: valid.RepoCfg{
 				Version: 3,
@@ -971,6 +997,14 @@ workflows:
 								},
 							},
 						},
+						StateRm: valid.Stage{
+							Steps: []valid.Step{
+								{
+									StepName:   "run",
+									RunCommand: "echo apply \"arg 4\"",
+								},
+							},
+						},
 					},
 				},
 			},
@@ -1003,6 +1037,11 @@ workflows:
      - env:
          name: env_name
          value: env_value
+    state_rm:
+      steps:
+      - env:
+          name: env_name
+          value: env_value
 `,
 			exp: valid.RepoCfg{
 				Version: 3,
@@ -1055,6 +1094,15 @@ workflows:
 								},
 							},
 						},
+						StateRm: valid.Stage{
+							Steps: []valid.Step{
+								{
+									StepName:    "env",
+									EnvVarName:  "env_name",
+									EnvVarValue: "env_value",
+								},
+							},
+						},
 					},
 				},
 			},
@@ -1194,6 +1242,17 @@ func TestParseGlobalCfg(t *testing.T) {
 				},
 			},
 		},
+		StateRm: valid.Stage{
+			Steps: []valid.Step{
+				{
+					StepName:   "run",
+					RunCommand: "custom command",
+				},
+				{
+					StepName: "state_rm",
+				},
+			},
+		},
 	}

 	conftestVersion, _ := version.NewVersion("v1.0.0")
@@ -1291,6 +1350,7 @@ workflows:
    plan:
    policy_check:
    import:
+    state_rm:
 `,
 			exp: valid.GlobalCfg{
 				Repos: defaultCfg.Repos,
@@ -1312,6 +1372,8 @@ workflows:
      steps:
    import:
      steps:
+    state_rm:
+      steps:
 `,
 			exp: valid.GlobalCfg{
 				Repos: defaultCfg.Repos,
@@ -1362,6 +1424,10 @@ workflows:
      steps:
      - run: custom command
      - import
+    state_rm:
+      steps:
+      - run: custom command
+      - state_rm
 policies:
  conftest_version: v1.0.0
  policy_sets:
@@ -1454,6 +1520,8 @@ workflows:
      steps: []
    import:
      steps: []
+    state_rm:
+      steps: []
 `,
 			exp: valid.GlobalCfg{
 				Repos: []valid.Repo{
@@ -1481,6 +1549,9 @@ workflows:
 							Import: valid.Stage{
 								Steps: nil,
 							},
+							StateRm: valid.Stage{
+								Steps: nil,
+							},
 						},
 						AllowedWorkflows:          []string{},
 						AllowedOverrides:          []string{},
@@ -1599,6 +1670,14 @@ func TestParserValidator_ParseGlobalCfgJSON(t *testing.T) {
 				},
 			},
 		},
+		StateRm: valid.Stage{
+			Steps: []valid.Step{
+				{
+					StepName:   "run",
+					RunCommand: "custom state_rm",
+				},
+			},
+		},
 	}

 	conftestVersion, _ := version.NewVersion("v1.0.0")
@@ -1661,6 +1740,11 @@ func TestParserValidator_ParseGlobalCfgJSON(t *testing.T) {
        "steps": [
          {"run": "custom import"}
        ]
+      },
+      "state_rm": {
+        "steps": [
+          {"run": "custom state_rm"}
+        ]
      }
    }
  },
@@ -1833,5 +1917,6 @@ func defaultWorkflow(name string) valid.Workflow {
 		Plan:        valid.DefaultPlanStage,
 		PolicyCheck: valid.DefaultPolicyCheckStage,
 		Import:      valid.DefaultImportStage,
+		StateRm:     valid.DefaultStateRmStage,
 	}
 }
--- a/server/core/config/raw/repo_cfg_test.go
+++ b/server/core/config/raw/repo_cfg_test.go
@@ -308,6 +308,7 @@ func TestConfig_ToValid(t *testing.T) {
 						Apply:       nil,
 						PolicyCheck: nil,
 						Import:      nil,
+						StateRm:     nil,
 					},
 				},
 			},
@@ -322,6 +323,7 @@ func TestConfig_ToValid(t *testing.T) {
 						PolicyCheck: valid.DefaultPolicyCheckStage,
 						Apply:       valid.DefaultApplyStage,
 						Import:      valid.DefaultImportStage,
+						StateRm:     valid.DefaultStateRmStage,
 					},
 				},
 			},
@@ -362,6 +364,13 @@ func TestConfig_ToValid(t *testing.T) {
 								},
 							},
 						},
+						StateRm: &raw.Stage{
+							Steps: []raw.Step{
+								{
+									Key: String("state_rm"),
+								},
+							},
+						},
 					},
 				},
 				Projects: []raw.Project{
@@ -405,6 +414,13 @@ func TestConfig_ToValid(t *testing.T) {
 								},
 							},
 						},
+						StateRm: valid.Stage{
+							Steps: []valid.Step{
+								{
+									StepName: "state_rm",
+								},
+							},
+						},
 					},
 				},
 				Projects: []valid.Project{
--- a/server/core/config/raw/step.go
+++ b/server/core/config/raw/step.go
@@ -25,6 +25,7 @@ const (
 	EnvStepName         = "env"
 	MultiEnvStepName    = "multienv"
 	ImportStepName      = "import"
+	StateRmStepName     = "state_rm"
 )

 // Step represents a single action/command to perform. In YAML, it can be set as
@@ -90,7 +91,8 @@ func (s Step) validStepName(stepName string) bool {
 		stepName == MultiEnvStepName ||
 		stepName == ShowStepName ||
 		stepName == PolicyCheckStepName ||
-		stepName == ImportStepName
+		stepName == ImportStepName ||
+		stepName == StateRmStepName
 }

 func (s Step) Validate() error {
--- a/server/core/config/raw/workflow.go
+++ b/server/core/config/raw/workflow.go
@@ -10,6 +10,7 @@ type Workflow struct {
 	Plan        *Stage `yaml:"plan,omitempty" json:"plan,omitempty"`
 	PolicyCheck *Stage `yaml:"policy_check,omitempty" json:"policy_check,omitempty"`
 	Import      *Stage `yaml:"import,omitempty" json:"import,omitempty"`
+	StateRm     *Stage `yaml:"state_rm,omitempty" json:"state_rm,omitempty"`
 }

 func (w Workflow) Validate() error {
@@ -18,6 +19,7 @@ func (w Workflow) Validate() error {
 		validation.Field(&w.Plan),
 		validation.Field(&w.PolicyCheck),
 		validation.Field(&w.Import),
+		validation.Field(&w.StateRm),
 	)
 }

@@ -38,6 +40,7 @@ func (w Workflow) ToValid(name string) valid.Workflow {
 	v.Plan = w.toValidStage(w.Plan, valid.DefaultPlanStage)
 	v.PolicyCheck = w.toValidStage(w.PolicyCheck, valid.DefaultPolicyCheckStage)
 	v.Import = w.toValidStage(w.Import, valid.DefaultImportStage)
+	v.StateRm = w.toValidStage(w.StateRm, valid.DefaultStateRmStage)

 	return v
 }
--- a/server/core/config/raw/workflow_test.go
+++ b/server/core/config/raw/workflow_test.go
@@ -149,6 +149,7 @@ func TestWorkflow_ToValid(t *testing.T) {
 				Plan:        valid.DefaultPlanStage,
 				PolicyCheck: valid.DefaultPolicyCheckStage,
 				Import:      valid.DefaultImportStage,
+				StateRm:     valid.DefaultStateRmStage,
 			},
 		},
 		{
@@ -182,6 +183,13 @@ func TestWorkflow_ToValid(t *testing.T) {
 						},
 					},
 				},
+				StateRm: &raw.Stage{
+					Steps: []raw.Step{
+						{
+							Key: String("state_rm"),
+						},
+					},
+				},
 			},
 			exp: valid.Workflow{
 				Apply: valid.Stage{
@@ -212,6 +220,13 @@ func TestWorkflow_ToValid(t *testing.T) {
 						},
 					},
 				},
+				StateRm: valid.Stage{
+					Steps: []valid.Step{
+						{
+							StepName: "state_rm",
+						},
+					},
+				},
 			},
 		},
 	}
--- a/server/core/config/valid/global_cfg.go
+++ b/server/core/config/valid/global_cfg.go
@@ -149,6 +149,18 @@ var DefaultImportStage = Stage{
 	},
 }

+// DefaultStateRmStage is the Atlantis default state_rm stage.
+var DefaultStateRmStage = Stage{
+	Steps: []Step{
+		{
+			StepName: "init",
+		},
+		{
+			StepName: "state_rm",
+		},
+	},
+}
+
 // Deprecated: use NewGlobalCfgFromArgs
 func NewGlobalCfgWithHooks(allowRepoCfg bool, mergeableReq bool, approvedReq bool, unDivergedReq bool, preWorkflowHooks []*WorkflowHook, postWorkflowHooks []*WorkflowHook) GlobalCfg {
 	return NewGlobalCfgFromArgs(GlobalCfgArgs{
@@ -194,6 +206,7 @@ func NewGlobalCfgFromArgs(args GlobalCfgArgs) GlobalCfg {
 		Plan:        DefaultPlanStage,
 		PolicyCheck: DefaultPolicyCheckStage,
 		Import:      DefaultImportStage,
+		StateRm:     DefaultStateRmStage,
 	}
 	// Must construct slices here instead of using a `var` declaration because
 	// we treat nil slices differently.
--- a/server/core/config/valid/global_cfg_test.go
+++ b/server/core/config/valid/global_cfg_test.go
@@ -55,6 +55,16 @@ func TestNewGlobalCfg(t *testing.T) {
 				},
 			},
 		},
+		StateRm: valid.Stage{
+			Steps: []valid.Step{
+				{
+					StepName: "init",
+				},
+				{
+					StepName: "state_rm",
+				},
+			},
+		},
 	}
 	baseCfg := valid.GlobalCfg{
 		Repos: []valid.Repo{
@@ -631,6 +641,7 @@ policies:
 					Plan:        valid.DefaultPlanStage,
 					PolicyCheck: valid.DefaultPolicyCheckStage,
 					Import:      valid.DefaultImportStage,
+					StateRm:     valid.DefaultStateRmStage,
 				},
 				PolicySets: valid.PolicySets{
 					Version: nil,
@@ -675,6 +686,7 @@ policies:
 					Plan:        valid.DefaultPlanStage,
 					PolicyCheck: valid.DefaultPolicyCheckStage,
 					Import:      valid.DefaultImportStage,
+					StateRm:     valid.DefaultStateRmStage,
 				},
 				PolicySets: valid.PolicySets{
 					Version: version,
@@ -736,6 +748,7 @@ func TestGlobalCfg_MergeProjectCfg(t *testing.T) {
 		PolicyCheck: valid.DefaultPolicyCheckStage,
 		Plan:        valid.DefaultPlanStage,
 		Import:      valid.DefaultImportStage,
+		StateRm:     valid.DefaultStateRmStage,
 	}
 	cases := map[string]struct {
 		gCfg          string
@@ -774,7 +787,8 @@ workflows:
 							},
 						},
 					},
-					Import: valid.DefaultImportStage,
+					Import:  valid.DefaultImportStage,
+					StateRm: valid.DefaultStateRmStage,
 				},
 				RepoRelDir:      ".",
 				Workspace:       "default",
--- a/server/core/config/valid/repo_cfg.go
+++ b/server/core/config/valid/repo_cfg.go
@@ -170,4 +170,5 @@ type Workflow struct {
 	Plan        Stage
 	PolicyCheck Stage
 	Import      Stage
+	StateRm     Stage
 }
--- a/server/core/runtime/import_step_runner_test.go
+++ b/server/core/runtime/import_step_runner_test.go
@@ -8,9 +8,9 @@ import (

 	"github.com/hashicorp/go-version"
 	. "github.com/petergtz/pegomock"
-	"github.com/runatlantis/atlantis/server/core/runtime/mocks/matchers"
+	runtimematchers "github.com/runatlantis/atlantis/server/core/runtime/mocks/matchers"
 	"github.com/runatlantis/atlantis/server/core/terraform/mocks"
-	matchers2 "github.com/runatlantis/atlantis/server/core/terraform/mocks/matchers"
+	tfmatchers "github.com/runatlantis/atlantis/server/core/terraform/mocks/matchers"
 	"github.com/runatlantis/atlantis/server/events/command"
 	"github.com/runatlantis/atlantis/server/logging"
 	. "github.com/runatlantis/atlantis/testing"
@@ -35,7 +35,7 @@ func TestImportStepRunner_Run_Success(t *testing.T) {
 	tfVersion, _ := version.NewVersion("0.15.0")
 	s := NewImportStepRunner(terraform, tfVersion)

-	When(terraform.RunCommandWithVersion(matchers.AnyCommandProjectContext(), AnyString(), AnyStringSlice(), matchers2.AnyMapOfStringToString(), matchers2.AnyPtrToGoVersionVersion(), AnyString())).
+	When(terraform.RunCommandWithVersion(runtimematchers.AnyCommandProjectContext(), AnyString(), AnyStringSlice(), tfmatchers.AnyMapOfStringToString(), tfmatchers.AnyPtrToGoVersionVersion(), AnyString())).
 		ThenReturn("output", nil)
 	output, err := s.Run(context, []string{}, tmpDir, map[string]string(nil))
 	Ok(t, err)
@@ -65,7 +65,7 @@ func TestImportStepRunner_Run_Workspace(t *testing.T) {
 	tfVersion, _ := version.NewVersion("0.15.0")
 	s := NewImportStepRunner(terraform, tfVersion)

-	When(terraform.RunCommandWithVersion(matchers.AnyCommandProjectContext(), AnyString(), AnyStringSlice(), matchers2.AnyMapOfStringToString(), matchers2.AnyPtrToGoVersionVersion(), AnyString())).
+	When(terraform.RunCommandWithVersion(runtimematchers.AnyCommandProjectContext(), AnyString(), AnyStringSlice(), runtimematchers.AnyMapOfStringToString(), runtimematchers.AnyPtrToGoVersionVersion(), AnyString())).
 		ThenReturn("output", nil)
 	output, err := s.Run(context, []string{}, tmpDir, map[string]string(nil))
 	Ok(t, err)
--- a/server/core/runtime/state_rm_step_runner.go
+++ b/server/core/runtime/state_rm_step_runner.go
@@ -0,0 +1,46 @@
+package runtime
+
+import (
+	"os"
+	"path/filepath"
+
+	version "github.com/hashicorp/go-version"
+	"github.com/runatlantis/atlantis/server/events/command"
+)
+
+type stateRmStepRunner struct {
+	terraformExecutor TerraformExec
+	defaultTFVersion  *version.Version
+}
+
+func NewStateRmStepRunner(terraformExecutor TerraformExec, defaultTfVersion *version.Version) Runner {
+	runner := &stateRmStepRunner{
+		terraformExecutor: terraformExecutor,
+		defaultTFVersion:  defaultTfVersion,
+	}
+	return NewWorkspaceStepRunnerDelegate(terraformExecutor, defaultTfVersion, runner)
+}
+
+func (p *stateRmStepRunner) Run(ctx command.ProjectContext, extraArgs []string, path string, envs map[string]string) (string, error) {
+	tfVersion := p.defaultTFVersion
+	if ctx.TerraformVersion != nil {
+		tfVersion = ctx.TerraformVersion
+	}
+
+	stateRmCmd := []string{"state", "rm"}
+	stateRmCmd = append(stateRmCmd, extraArgs...)
+	stateRmCmd = append(stateRmCmd, ctx.EscapedCommentArgs...)
+	out, err := p.terraformExecutor.RunCommandWithVersion(ctx, filepath.Clean(path), stateRmCmd, envs, tfVersion, ctx.Workspace)
+
+	// If the state rm was successful and a plan file exists, delete the plan.
+	planPath := filepath.Join(path, GetPlanFilename(ctx.Workspace, ctx.ProjectName))
+	if err == nil {
+		if _, planPathErr := os.Stat(planPath); !os.IsNotExist(planPathErr) {
+			ctx.Log.Info("state rm successful, deleting planfile")
+			if removeErr := os.Remove(planPath); removeErr != nil {
+				ctx.Log.Warn("failed to delete planfile after successful state rm: %s", removeErr)
+			}
+		}
+	}
+	return out, err
+}
--- a/server/core/runtime/state_rm_step_runner_test.go
+++ b/server/core/runtime/state_rm_step_runner_test.go
@@ -0,0 +1,84 @@
+package runtime
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/hashicorp/go-version"
+	. "github.com/petergtz/pegomock"
+	runtimematchers "github.com/runatlantis/atlantis/server/core/runtime/mocks/matchers"
+	"github.com/runatlantis/atlantis/server/core/terraform/mocks"
+	tfmatchers "github.com/runatlantis/atlantis/server/core/terraform/mocks/matchers"
+	"github.com/runatlantis/atlantis/server/events/command"
+	"github.com/runatlantis/atlantis/server/logging"
+	. "github.com/runatlantis/atlantis/testing"
+)
+
+func TestStateRmStepRunner_Run_Success(t *testing.T) {
+	logger := logging.NewNoopLogger(t)
+	workspace := "default"
+	tmpDir := t.TempDir()
+	planPath := filepath.Join(tmpDir, fmt.Sprintf("%s.tfplan", workspace))
+	err := os.WriteFile(planPath, nil, 0600)
+	Ok(t, err)
+
+	context := command.ProjectContext{
+		Log:                logger,
+		EscapedCommentArgs: []string{"-lock=false", "addr1", "addr2", "addr3"},
+		Workspace:          workspace,
+	}
+
+	RegisterMockTestingT(t)
+	terraform := mocks.NewMockClient()
+	tfVersion, _ := version.NewVersion("0.15.0")
+	s := NewStateRmStepRunner(terraform, tfVersion)
+
+	When(terraform.RunCommandWithVersion(runtimematchers.AnyCommandProjectContext(), AnyString(), AnyStringSlice(), tfmatchers.AnyMapOfStringToString(), tfmatchers.AnyPtrToGoVersionVersion(), AnyString())).
+		ThenReturn("output", nil)
+	output, err := s.Run(context, []string{}, tmpDir, map[string]string(nil))
+	Ok(t, err)
+	Equals(t, "output", output)
+	commands := []string{"state", "rm", "-lock=false", "addr1", "addr2", "addr3"}
+	terraform.VerifyWasCalledOnce().RunCommandWithVersion(context, tmpDir, commands, map[string]string(nil), tfVersion, "default")
+	_, err = os.Stat(planPath)
+	Assert(t, os.IsNotExist(err), "planfile should be deleted")
+}
+
+func TestStateRmStepRunner_Run_Workspace(t *testing.T) {
+	logger := logging.NewNoopLogger(t)
+	workspace := "something"
+	tmpDir := t.TempDir()
+	planPath := filepath.Join(tmpDir, fmt.Sprintf("%s.tfplan", workspace))
+	err := os.WriteFile(planPath, nil, 0600)
+	Ok(t, err)
+
+	context := command.ProjectContext{
+		Log:                logger,
+		EscapedCommentArgs: []string{"-lock=false", "addr1", "addr2", "addr3"},
+		Workspace:          workspace,
+	}
+
+	RegisterMockTestingT(t)
+	terraform := mocks.NewMockClient()
+	tfVersion, _ := version.NewVersion("0.15.0")
+	s := NewStateRmStepRunner(terraform, tfVersion)
+
+	When(terraform.RunCommandWithVersion(runtimematchers.AnyCommandProjectContext(), AnyString(), AnyStringSlice(), runtimematchers.AnyMapOfStringToString(), runtimematchers.AnyPtrToGoVersionVersion(), AnyString())).
+		ThenReturn("output", nil)
+	output, err := s.Run(context, []string{}, tmpDir, map[string]string(nil))
+	Ok(t, err)
+	Equals(t, "output", output)
+
+	// switch workspace
+	terraform.VerifyWasCalledOnce().RunCommandWithVersion(context, tmpDir, []string{"workspace", "show"}, map[string]string(nil), tfVersion, workspace)
+	terraform.VerifyWasCalledOnce().RunCommandWithVersion(context, tmpDir, []string{"workspace", "select", workspace}, map[string]string(nil), tfVersion, workspace)
+
+	// exec state rm
+	commands := []string{"state", "rm", "-lock=false", "addr1", "addr2", "addr3"}
+	terraform.VerifyWasCalledOnce().RunCommandWithVersion(context, tmpDir, commands, map[string]string(nil), tfVersion, workspace)
+
+	_, err = os.Stat(planPath)
+	Assert(t, os.IsNotExist(err), "planfile should be deleted")
+}
--- a/server/events/command/name.go
+++ b/server/events/command/name.go
@@ -28,9 +28,16 @@ const (
 	Version
 	// Import is a command to run terraform import
 	Import
+	// State is a command to run terraform state rm
+	State
 	// Adding more? Don't forget to update String() below
 )

+type ArgCount struct {
+	Min int
+	Max int
+}
+
 // AllCommentCommands are list of commands that can be run from a comment.
 var AllCommentCommands = []Name{
 	Version,
@@ -39,6 +46,7 @@ var AllCommentCommands = []Name{
 	Unlock,
 	ApprovePolicies,
 	Import,
+	State,
 }

 // TitleString returns the string representation in title form.
@@ -64,6 +72,8 @@ func (c Name) String() string {
 		return "version"
 	case Import:
 		return "import"
+	case State:
+		return "state"
 	}
 	return ""
 }
@@ -73,11 +83,53 @@ func (c Name) DefaultUsage() string {
 	switch c {
 	case Import:
 		return "import ADDRESS ID"
+	case State:
+		return "state [rm ADDRESS...]"
 	default:
 		return c.String()
 	}
 }

+// SubCommands returns the list of sub commands for the command
+func (c Name) SubCommands() []string {
+	switch c {
+	case State:
+		return []string{"rm"}
+	default:
+		return nil
+	}
+}
+
+// CommandArgCount returns the number of required arguments for the command
+func (c Name) CommandArgCount(subCommand string) (*ArgCount, error) {
+	switch c {
+	case Import:
+		return &ArgCount{2, 2}, nil // "atlantis import ADDRESS ID"
+	case State:
+		if subCommand == "rm" {
+			return &ArgCount{1, -1}, nil // "atlantis state rm ADDRESS..."
+		}
+		return nil, fmt.Errorf("command arg count unknown sub command: %s", subCommand)
+	default:
+		return &ArgCount{0, 0}, nil // other command doesn't require any args
+	}
+}
+
+// IsMatchCount returns true if the number of arguments matches the requirement
+func (a ArgCount) IsMatchCount(count int) bool {
+	if a.Min != -1 {
+		if count < a.Min {
+			return false
+		}
+	}
+	if a.Max != -1 {
+		if count > a.Max {
+			return false
+		}
+	}
+	return true
+}
+
 // ParseCommandName parses raw name into a command name.
 func ParseCommandName(name string) (Name, error) {
 	switch name {
@@ -95,6 +147,8 @@ func ParseCommandName(name string) (Name, error) {
 		return Version, nil
 	case "import":
 		return Import, nil
+	case "state":
+		return State, nil
 	}
 	return -1, fmt.Errorf("unknown command name: %s", name)
 }
--- a/server/events/command/name_test.go
+++ b/server/events/command/name_test.go
@@ -1,6 +1,9 @@
 package command_test

 import (
+	"fmt"
+	"math"
+	"reflect"
 	"testing"

 	"github.com/runatlantis/atlantis/server/events/command"
@@ -36,6 +39,7 @@ func TestName_String(t *testing.T) {
 		{command.ApprovePolicies, "approve_policies"},
 		{command.Version, "version"},
 		{command.Import, "import"},
+		{command.State, "state"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.want, func(t *testing.T) {
@@ -58,6 +62,7 @@ func TestName_DefaultUsage(t *testing.T) {
 		{command.ApprovePolicies, "approve_policies"},
 		{command.Version, "version"},
 		{command.Import, "import ADDRESS ID"},
+		{command.State, "state [rm ADDRESS...]"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.c.String(), func(t *testing.T) {
@@ -68,6 +73,98 @@ func TestName_DefaultUsage(t *testing.T) {
 	}
 }

+func TestName_SubCommands(t *testing.T) {
+	tests := []struct {
+		c    command.Name
+		want []string
+	}{
+		{c: command.Apply},
+		{c: command.Plan},
+		{c: command.Unlock},
+		{c: command.PolicyCheck},
+		{c: command.ApprovePolicies},
+		{c: command.Version},
+		{c: command.Import},
+		{c: command.State, want: []string{"rm"}},
+	}
+	for _, tt := range tests {
+		t.Run(tt.c.String(), func(t *testing.T) {
+			if got := tt.c.SubCommands(); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("SubCommands() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestName_CommandArgCount(t *testing.T) {
+	tests := []struct {
+		c          command.Name
+		subCommand string
+		want       *command.ArgCount
+		wantErr    bool
+	}{
+		{c: command.Apply, want: &command.ArgCount{}},
+		{c: command.Plan, want: &command.ArgCount{}},
+		{c: command.Unlock, want: &command.ArgCount{}},
+		{c: command.PolicyCheck, want: &command.ArgCount{}},
+		{c: command.ApprovePolicies, want: &command.ArgCount{}},
+		{c: command.Version, want: &command.ArgCount{}},
+		{c: command.Import, want: &command.ArgCount{Min: 2, Max: 2}},
+		{c: command.State, subCommand: "rm", want: &command.ArgCount{Min: 1, Max: -1}},
+		{c: command.State, subCommand: "unknown", wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(fmt.Sprintf("%s %s", tt.c, tt.subCommand), func(t *testing.T) {
+			got, err := tt.c.CommandArgCount(tt.subCommand)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("CommandArgCount() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("CommandArgCount() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestArgCount_IsMatchCount(t *testing.T) {
+	type fields struct {
+		Min int
+		Max int
+	}
+	tests := []struct {
+		name   string
+		fields fields
+		count  int
+		want   bool
+	}{
+		{name: "[0,0] success", fields: fields{Min: 0, Max: 0}, count: 0, want: true},
+		{name: "[0,0] failure", fields: fields{Min: 0, Max: 0}, count: 1, want: false},
+		{name: "[1,1] success", fields: fields{Min: 1, Max: 1}, count: 1, want: true},
+		{name: "[1,1] failure1", fields: fields{Min: 1, Max: 1}, count: 0, want: false},
+		{name: "[1,1] failure2", fields: fields{Min: 1, Max: 1}, count: 2, want: false},
+		{name: "[-inf,1] success1", fields: fields{Min: -1, Max: 1}, count: 0, want: true},
+		{name: "[-inf,1] success2", fields: fields{Min: -1, Max: 1}, count: 1, want: true},
+		{name: "[-inf,1] failure", fields: fields{Min: -1, Max: 1}, count: 2, want: false},
+		{name: "[1,inf] success1", fields: fields{Min: 1, Max: -1}, count: 1, want: true},
+		{name: "[1,inf] success2", fields: fields{Min: 1, Max: -1}, count: math.MaxInt, want: true},
+		{name: "[1,inf] failure", fields: fields{Min: 1, Max: -1}, count: 0, want: false},
+		{name: "[-inf,inf] success", fields: fields{Min: -1, Max: -1}, count: 0, want: true},
+		{name: "[-inf,inf] success", fields: fields{Min: -1, Max: -1}, count: math.MaxInt, want: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			a := command.ArgCount{
+				Min: tt.fields.Min,
+				Max: tt.fields.Max,
+			}
+			if got := a.IsMatchCount(tt.count); got != tt.want {
+				t.Errorf("IsMatchCount() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func TestParseCommandName(t *testing.T) {
 	tests := []struct {
 		exp  command.Name
@@ -80,6 +177,7 @@ func TestParseCommandName(t *testing.T) {
 		{command.ApprovePolicies, "approve_policies"},
 		{command.Version, "version"},
 		{command.Import, "import"},
+		{command.State, "state"},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
--- a/server/events/command/project_result.go
+++ b/server/events/command/project_result.go
@@ -7,6 +7,7 @@ import (
 // ProjectResult is the result of executing a plan/policy_check/apply for a specific project.
 type ProjectResult struct {
 	Command            Name
+	SubCommand         string
 	RepoRelDir         string
 	Workspace          string
 	Error              error
@@ -16,6 +17,7 @@ type ProjectResult struct {
 	ApplySuccess       string
 	VersionSuccess     string
 	ImportSuccess      *models.ImportSuccess
+	StateRmSuccess     *models.StateRmSuccess
 	ProjectName        string
 }

--- a/server/events/comment_parser.go
+++ b/server/events/comment_parser.go
@@ -27,6 +27,7 @@ import (
 	"github.com/runatlantis/atlantis/server/events/command"
 	"github.com/runatlantis/atlantis/server/events/models"
 	"github.com/spf13/pflag"
+	"golang.org/x/exp/slices"
 )

 const (
@@ -206,7 +207,6 @@ func (e *CommentParser) Parse(rawComment string, vcsHost models.VCSHostType) Com
 	var verbose, autoMergeDisabled bool
 	var flagSet *pflag.FlagSet
 	var name command.Name
-	var requiredCommandArgCount int

 	// Set up the flag parsing depending on the command.
 	switch cmd {
@@ -251,47 +251,22 @@ func (e *CommentParser) Parse(rawComment string, vcsHost models.VCSHostType) Com
 		flagSet.StringVarP(&dir, dirFlagLong, dirFlagShort, "", "Which directory to run import in relative to root of repo, ex. 'child/dir'.")
 		flagSet.StringVarP(&project, projectFlagLong, projectFlagShort, "", "Which project to run import for. Refers to the name of the project configured in a repo config file. Cannot be used at same time as workspace or dir flags.")
 		flagSet.BoolVarP(&verbose, verboseFlagLong, verboseFlagShort, false, "Append Atlantis log to comment.")
-		requiredCommandArgCount = 2 // import requires `atlantis import ADDRESS ID` args
+	case command.State.String():
+		name = command.State
+		flagSet = pflag.NewFlagSet(command.State.String(), pflag.ContinueOnError)
+		flagSet.SetOutput(io.Discard)
+		flagSet.StringVarP(&workspace, workspaceFlagLong, workspaceFlagShort, "", "Switch to this Terraform workspace before processing tfstate.")
+		flagSet.StringVarP(&dir, dirFlagLong, dirFlagShort, "", "Which directory to run state command in relative to root of repo, ex. 'child/dir'.")
+		flagSet.StringVarP(&project, projectFlagLong, projectFlagShort, "", "Which project to run state command for. Refers to the name of the project configured in a repo config file. Cannot be used at same time as workspace or dir flags.")
+		flagSet.BoolVarP(&verbose, verboseFlagLong, verboseFlagShort, false, "Append Atlantis log to comment.")
 	default:
 		return CommentParseResult{CommentResponse: fmt.Sprintf("Error: unknown command %q – this is a bug", cmd)}
 	}

-	// Now parse the flags.
-	// It's safe to use [2:] because we know there's at least 2 elements in args.
-	err = flagSet.Parse(args[2:])
-	if err == pflag.ErrHelp {
-		return CommentParseResult{CommentResponse: fmt.Sprintf("```\nUsage of %s:\n%s\n```", name.DefaultUsage(), flagSet.FlagUsagesWrapped(usagesCols))}
+	subName, extraArgs, errResult := e.parseArgs(name, args, flagSet)
+	if errResult != "" {
+		return CommentParseResult{CommentResponse: errResult}
 	}
-	if err != nil {
-		if cmd == command.Unlock.String() {
-			return CommentParseResult{CommentResponse: fmt.Sprintf(UnlockUsage, e.ExecutableName)}
-		}
-		return CommentParseResult{CommentResponse: e.errMarkdown(err.Error(), cmd, flagSet)}
-	}
-
-	var commandArgs []string // commandArgs are the arguments that are passed before `--` without any parameter flags.
-	if flagSet.ArgsLenAtDash() == -1 {
-		commandArgs = flagSet.Args()
-	} else {
-		commandArgs = flagSet.Args()[0:flagSet.ArgsLenAtDash()]
-	}
-	if len(commandArgs) != requiredCommandArgCount {
-		if requiredCommandArgCount > 0 {
-			return CommentParseResult{CommentResponse: e.errMarkdown(fmt.Sprintf("invalid argument count – %s", strings.Join(commandArgs, " ")), name.DefaultUsage(), flagSet)}
-		}
-		return CommentParseResult{CommentResponse: e.errMarkdown(fmt.Sprintf("unknown argument(s) – %s", strings.Join(commandArgs, " ")), name.DefaultUsage(), flagSet)}
-	}
-
-	var extraArgs []string // command extra_args
-	if flagSet.ArgsLenAtDash() != -1 {
-		extraArgs = append(extraArgs, flagSet.Args()[flagSet.ArgsLenAtDash():]...)
-	}
-	// pass commandArgs into extraArgs after extra args.
-	// - after comment_parser, we will use extra_args only.
-	// - terraform command args accept after options like followings
-	//   - from: `atlantis import ADDRESS ID -- -var foo=bar
-	//   - to: `terraform import -var foo=bar ADDRESS ID`
-	extraArgs = append(extraArgs, commandArgs...)

 	dir, err = e.validateDir(dir)
 	if err != nil {
@@ -316,10 +291,73 @@ func (e *CommentParser) Parse(rawComment string, vcsHost models.VCSHostType) Com
 	}

 	return CommentParseResult{
-		Command: NewCommentCommand(dir, extraArgs, name, verbose, autoMergeDisabled, workspace, project),
+		Command: NewCommentCommand(dir, extraArgs, name, subName, verbose, autoMergeDisabled, workspace, project),
 	}
 }

+func (e *CommentParser) parseArgs(name command.Name, args []string, flagSet *pflag.FlagSet) (string, []string, string) {
+	// Now parse the flags.
+	// It's safe to use [2:] because we know there's at least 2 elements in args.
+	err := flagSet.Parse(args[2:])
+	if err == pflag.ErrHelp {
+		return "", nil, fmt.Sprintf("```\nUsage of %s:\n%s\n```", name.DefaultUsage(), flagSet.FlagUsagesWrapped(usagesCols))
+	}
+	if err != nil {
+		if name == command.Unlock {
+			return "", nil, fmt.Sprintf(UnlockUsage, e.ExecutableName)
+		}
+		return "", nil, e.errMarkdown(err.Error(), name.String(), flagSet)
+	}
+
+	var commandArgs []string // commandArgs are the arguments that are passed before `--` without any parameter flags.
+	if flagSet.ArgsLenAtDash() == -1 {
+		commandArgs = flagSet.Args()
+	} else {
+		commandArgs = flagSet.Args()[0:flagSet.ArgsLenAtDash()]
+	}
+
+	// If command require subcommand, get it from command args
+	var subCommand string
+	availableSubCommands := name.SubCommands()
+	if len(availableSubCommands) > 0 { // command requires a subcommand
+		if len(commandArgs) < 1 {
+			return "", nil, e.errMarkdown("subcommand required", name.String(), flagSet)
+		}
+		subCommand, commandArgs = commandArgs[0], commandArgs[1:]
+		isAvailableSubCommand := slices.Contains(availableSubCommands, subCommand)
+		if !isAvailableSubCommand {
+			errMsg := fmt.Sprintf("invalid subcommand %s (not %s)", subCommand, strings.Join(availableSubCommands, ", "))
+			return "", nil, e.errMarkdown(errMsg, name.String(), flagSet)
+		}
+	}
+
+	// check command args count requirements
+	commandArgCount, err := name.CommandArgCount(subCommand)
+	if err != nil {
+		return "", nil, e.errMarkdown(err.Error(), name.String(), flagSet)
+	}
+	if !commandArgCount.IsMatchCount(len(commandArgs)) {
+		return "", nil, e.errMarkdown(fmt.Sprintf("unknown argument(s) – %s", strings.Join(commandArgs, " ")), name.DefaultUsage(), flagSet)
+	}
+
+	var extraArgs []string // command extra_args
+	if flagSet.ArgsLenAtDash() != -1 {
+		extraArgs = append(extraArgs, flagSet.Args()[flagSet.ArgsLenAtDash():]...)
+	}
+
+	// pass commandArgs into extraArgs after extra args.
+	// - after comment_parser, we will use extra_args only.
+	// - terraform command args accept after options like followings
+	//   - e.g.
+	//     - from: `atlantis import ADDRESS ID -- -var foo=bar
+	//     - to: `terraform import -var foo=bar ADDRESS ID`
+	//   - e.g.
+	//     - from: `atlantis state rm ADDRESS1 ADDRESS2 -- -var foo=bar
+	//     - to: `terraform state rm -var foo=bar ADDRESS1 ADDRESS2` (subcommand=rm)
+	extraArgs = append(extraArgs, commandArgs...)
+	return subCommand, extraArgs, ""
+}
+
 // BuildPlanComment builds a plan comment for the specified args.
 func (e *CommentParser) BuildPlanComment(repoRelDir string, workspace string, project string, commentArgs []string) string {
 	flags := e.buildFlags(repoRelDir, workspace, project, false)
@@ -424,6 +462,7 @@ func (e *CommentParser) HelpComment() string {
 		AllowUnlock          bool
 		AllowApprovePolicies bool
 		AllowImport          bool
+		AllowState           bool
 	}{
 		ExecutableName:       e.ExecutableName,
 		AllowVersion:         e.isAllowedCommand(command.Version.String()),
@@ -432,6 +471,7 @@ func (e *CommentParser) HelpComment() string {
 		AllowUnlock:          e.isAllowedCommand(command.Unlock.String()),
 		AllowApprovePolicies: e.isAllowedCommand(command.ApprovePolicies.String()),
 		AllowImport:          e.isAllowedCommand(command.Import.String()),
+		AllowState:           e.isAllowedCommand(command.State.String()),
 	}); err != nil {
 		return fmt.Sprintf("Failed to render template, this is a bug: %v", err)
 	}
@@ -483,8 +523,14 @@ Commands:
  version  Print the output of 'terraform version'
 {{- end }}
 {{- if .AllowImport }}
-  import   Runs 'terraform import' for the changes in this pull request.
-           To plan a specific project, use the -d, -w and -p flags.
+  import ADDRESS ID
+           Runs 'terraform import' for the passed address resource.
+           To import a specific project, use the -d, -w and -p flags.
+{{- end }}
+{{- if .AllowState }}
+  state rm ADDRESS...
+           Runs 'terraform state rm' for the passed address resource.
+           To remove a specific project resource, use the -d, -w and -p flags.
 {{- end }}
  help     View help.

--- a/server/events/comment_parser_test.go
+++ b/server/events/comment_parser_test.go
@@ -29,13 +29,7 @@ var commentParser = events.CommentParser{
 	GithubUser:     "github-user",
 	GitlabUser:     "gitlab-user",
 	ExecutableName: "atlantis",
-	AllowCommands: []command.Name{
-		command.Plan,
-		command.Apply,
-		command.Unlock,
-		command.ApprovePolicies,
-		command.Import,
-	},
+	AllowCommands:  command.AllCommentCommands,
 }

 func TestNewCommentParser(t *testing.T) {
@@ -151,82 +145,69 @@ func TestParse_HelpResponse(t *testing.T) {
 func TestParse_UnusedArguments(t *testing.T) {
 	t.Log("if there are unused flags we return an error")
 	cases := []struct {
-		Command    command.Name
-		Args       string
-		Unused     string
-		ExpMessage string
+		Command command.Name
+		Args    string
+		Unused  string
 	}{
 		{
 			command.Plan,
 			"-d . arg",
 			"arg",
-			"unknown argument(s)",
 		},
 		{
 			command.Plan,
 			"arg -d .",
 			"arg",
-			"unknown argument(s)",
 		},
 		{
 			command.Plan,
 			"arg",
 			"arg",
-			"unknown argument(s)",
 		},
 		{
 			command.Plan,
 			"arg arg2",
 			"arg arg2",
-			"unknown argument(s)",
 		},
 		{
 			command.Plan,
 			"-d . arg -w kjj arg2",
 			"arg arg2",
-			"unknown argument(s)",
 		},
 		{
 			command.Apply,
 			"-d . arg",
 			"arg",
-			"unknown argument(s)",
 		},
 		{
 			command.Apply,
 			"arg arg2",
 			"arg arg2",
-			"unknown argument(s)",
 		},
 		{
 			command.Apply,
 			"arg arg2 -- useful",
 			"arg arg2",
-			"unknown argument(s)",
 		},
 		{
 			command.Apply,
 			"arg arg2 --",
 			"arg arg2",
-			"unknown argument(s)",
 		},
 		{
 			command.ApprovePolicies,
 			"arg arg2 --",
 			"arg arg2",
-			"unknown argument(s)",
 		},
 		{
 			command.Import,
 			"arg --",
 			"arg",
-			"invalid argument count",
 		},
 		{
 			command.Import,
 			"arg1 arg2 arg3 --",
 			"arg1 arg2 arg3",
-			"invalid argument count",
 		},
 	}
 	for _, c := range cases {
@@ -244,7 +225,7 @@ func TestParse_UnusedArguments(t *testing.T) {
 			case command.Import:
 				usage = ImportUsage
 			}
-			Equals(t, fmt.Sprintf("```\nError: %s – %s.\n%s```", c.ExpMessage, c.Unused, usage), r.CommentResponse)
+			Equals(t, fmt.Sprintf("```\nError: unknown argument(s) – %s.\n%s```", c.Unused, usage), r.CommentResponse)
 		})
 	}
 }
@@ -319,6 +300,8 @@ func TestParse_SubcommandUsage(t *testing.T) {
 		{"atlantis approve_policies --help", "approve_policies"},
 		{"atlantis import -h", "import ADDRESS ID"},
 		{"atlantis import --help", "import ADDRESS ID"},
+		{"atlantis state -h", "state [rm ADDRESS...]"},
+		{"atlantis state --help", "state [rm ADDRESS...]"},
 	}
 	for _, c := range tests {
 		r := commentParser.Parse(c.input, models.Github)
@@ -357,6 +340,10 @@ func TestParse_InvalidFlags(t *testing.T) {
 			"atlantis import --abc",
 			"Error: unknown flag: --abc",
 		},
+		{
+			"atlantis state rm --abc",
+			"Error: unknown flag: --abc",
+		},
 	}
 	for _, c := range cases {
 		r := commentParser.Parse(c.comment, models.Github)
@@ -373,16 +360,20 @@ func TestParse_RelativeDirPath(t *testing.T) {
 		"atlantis plan -d ..",
 		"atlantis apply -d ..",
 		"atlantis import -d .. address id",
+		"atlantis state -d .. rm address",
 		// These won't return an error because we prepend with . when parsing.
 		//"atlantis plan -d /..",
 		//"atlantis apply -d /..",
 		//"atlantis import -d /.. address id",
+		//"atlantis state rm -d /.. address",
 		"atlantis plan -d ./..",
 		"atlantis apply -d ./..",
 		"atlantis import -d ./.. address id",
+		"atlantis state -d ./.. rm address",
 		"atlantis plan -d a/b/../../..",
 		"atlantis apply -d a/../..",
 		"atlantis import -d a/../.. address id",
+		"atlantis state -d a/../.. rm address id",
 	}
 	for _, c := range comments {
 		r := commentParser.Parse(c, models.Github)
@@ -427,15 +418,19 @@ func TestParse_InvalidWorkspace(t *testing.T) {
 		"atlantis plan -w ..",
 		"atlantis apply -w ..",
 		"atlantis import -w .. address id",
+		"atlantis import -w .. rm address",
 		"atlantis plan -w /",
 		"atlantis apply -w /",
 		"atlantis import -w / address id",
+		"atlantis state -w / rm address",
 		"atlantis plan -w ..abc",
 		"atlantis apply -w abc..",
 		"atlantis import -w abc.. address id",
+		"atlantis state -w abc.. rm address",
 		"atlantis plan -w abc..abc",
 		"atlantis apply -w ../../../etc/passwd",
 		"atlantis import -w ../../../etc/passwd address id",
+		"atlantis state -w ../../../etc/passwd rm address",
 	}
 	for _, c := range comments {
 		r := commentParser.Parse(c, models.Github)
@@ -681,7 +676,7 @@ func TestParse_Parsing(t *testing.T) {
 	}

 	for _, test := range cases {
-		for _, cmdName := range []string{"plan", "apply", "import 'some[\"addr\"]' id"} {
+		for _, cmdName := range []string{"plan", "apply", "import 'some[\"addr\"]' id", "state rm 'some[\"addr\"]'"} {
 			comment := fmt.Sprintf("atlantis %s %s", cmdName, test.flags)
 			t.Run(comment, func(t *testing.T) {
 				r := commentParser.Parse(comment, models.Github)
@@ -710,6 +705,15 @@ func TestParse_Parsing(t *testing.T) {
 					Assert(t, r.Command.Name == command.Import, "did not parse comment %q as import command", comment)
 					Assert(t, expExtraArgs == actExtraArgs, "exp extra args to equal %v but got %v for comment %q", expExtraArgs, actExtraArgs, comment)
 				}
+				if strings.HasPrefix(cmdName, "state rm") {
+					expExtraArgs := "some[\"addr\"]" // state rm use default args with `some["addr"]`
+					if test.expExtraArgs != "" {
+						expExtraArgs = fmt.Sprintf("%s %s", test.expExtraArgs, expExtraArgs)
+					}
+					Assert(t, r.Command.Name == command.State, "did not parse comment %q as state command", comment)
+					Assert(t, r.Command.SubName == "rm", "did not parse comment %q as state rm subcommand", comment)
+					Assert(t, expExtraArgs == actExtraArgs, "exp extra args to equal %v but got %v for comment %q", expExtraArgs, actExtraArgs, comment)
+				}
 			})
 		}
 	}
@@ -873,8 +877,12 @@ Commands:
  approve_policies
           Approves all current policy checking failures for the PR.
  version  Print the output of 'terraform version'
-  import   Runs 'terraform import' for the changes in this pull request.
-           To plan a specific project, use the -d, -w and -p flags.
+  import ADDRESS ID
+           Runs 'terraform import' for the passed address resource.
+           To import a specific project, use the -d, -w and -p flags.
+  state rm ADDRESS...
+           Runs 'terraform state rm' for the passed address resource.
+           To remove a specific project resource, use the -d, -w and -p flags.
  help     View help.

 Flags:
--- a/server/events/event_parser.go
+++ b/server/events/event_parser.go
@@ -38,6 +38,8 @@ const usagesCols = 90
 type PullCommand interface {
 	// CommandName is the name of the command we're running.
 	CommandName() command.Name
+	// SubCommandName is the subcommand name of the command we're running.
+	SubCommandName() string
 	// IsVerbose is true if the output of this command should be verbose.
 	IsVerbose() bool
 	// IsAutoplan is true if this is an autoplan command vs. a comment command.
@@ -53,6 +55,11 @@ func (c PolicyCheckCommand) CommandName() command.Name {
 	return command.PolicyCheck
 }

+// SubCommandName is a subcommand for policy_check.
+func (c PolicyCheckCommand) SubCommandName() string {
+	return ""
+}
+
 // IsVerbose is false for policy_check commands.
 func (c PolicyCheckCommand) IsVerbose() bool {
 	return false
@@ -72,6 +79,11 @@ func (c AutoplanCommand) CommandName() command.Name {
 	return command.Plan
 }

+// SubCommandName is a subcommand for auto plan.
+func (c AutoplanCommand) SubCommandName() string {
+	return ""
+}
+
 // IsVerbose is false for autoplan commands.
 func (c AutoplanCommand) IsVerbose() bool {
 	return false
@@ -92,6 +104,8 @@ type CommentCommand struct {
 	Flags []string
 	// Name is the name of the command the comment specified.
 	Name command.Name
+	// SubName is the name of the sub command the comment specified.
+	SubName string
 	// AutoMergeDisabled is true if the command should not automerge after apply.
 	AutoMergeDisabled bool
 	// Verbose is true if the command should output verbosely.
@@ -117,6 +131,11 @@ func (c CommentCommand) CommandName() command.Name {
 	return c.Name
 }

+// SubCommandName returns the name of this subcommand.
+func (c CommentCommand) SubCommandName() string {
+	return c.SubName
+}
+
 // IsVerbose is true if the command should give verbose output.
 func (c CommentCommand) IsVerbose() bool {
 	return c.Verbose
@@ -133,7 +152,7 @@ func (c CommentCommand) String() string {
 }

 // NewCommentCommand constructs a CommentCommand, setting all missing fields to defaults.
-func NewCommentCommand(repoRelDir string, flags []string, name command.Name, verbose, autoMergeDisabled bool, workspace string, project string) *CommentCommand {
+func NewCommentCommand(repoRelDir string, flags []string, name command.Name, subName string, verbose, autoMergeDisabled bool, workspace string, project string) *CommentCommand {
 	// If repoRelDir was empty we want to keep it that way to indicate that it
 	// wasn't specified in the comment.
 	if repoRelDir != "" {
@@ -146,6 +165,7 @@ func NewCommentCommand(repoRelDir string, flags []string, name command.Name, ver
 		RepoRelDir:        repoRelDir,
 		Flags:             flags,
 		Name:              name,
+		SubName:           subName,
 		Verbose:           verbose,
 		Workspace:         workspace,
 		AutoMergeDisabled: autoMergeDisabled,
--- a/server/events/event_parser_test.go
+++ b/server/events/event_parser_test.go
@@ -729,14 +729,14 @@ func TestNewCommand_CleansDir(t *testing.T) {

 	for _, c := range cases {
 		t.Run(c.RepoRelDir, func(t *testing.T) {
-			cmd := events.NewCommentCommand(c.RepoRelDir, nil, command.Plan, false, false, "workspace", "")
+			cmd := events.NewCommentCommand(c.RepoRelDir, nil, command.Plan, "", false, false, "workspace", "")
 			Equals(t, c.ExpDir, cmd.RepoRelDir)
 		})
 	}
 }

 func TestNewCommand_EmptyDirWorkspaceProject(t *testing.T) {
-	cmd := events.NewCommentCommand("", nil, command.Plan, false, false, "", "")
+	cmd := events.NewCommentCommand("", nil, command.Plan, "", false, false, "", "")
 	Equals(t, events.CommentCommand{
 		RepoRelDir:  "",
 		Flags:       nil,
@@ -748,7 +748,7 @@ func TestNewCommand_EmptyDirWorkspaceProject(t *testing.T) {
 }

 func TestNewCommand_AllFieldsSet(t *testing.T) {
-	cmd := events.NewCommentCommand("dir", []string{"a", "b"}, command.Plan, true, false, "workspace", "project")
+	cmd := events.NewCommentCommand("dir", []string{"a", "b"}, command.Plan, "", true, false, "workspace", "project")
 	Equals(t, events.CommentCommand{
 		Workspace:   "workspace",
 		RepoRelDir:  "dir",
--- a/server/events/instrumented_project_command_builder.go
+++ b/server/events/instrumented_project_command_builder.go
@@ -40,7 +40,7 @@ func (b *InstrumentedProjectCommandBuilder) BuildPlanCommands(ctx *command.Conte
 	)
 }

-func (b *InstrumentedProjectCommandBuilder) BuildVersionCommands(ctx *command.Context, comment *CommentCommand) ([]command.ProjectContext, error) {
+func (b *InstrumentedProjectCommandBuilder) BuildImportCommands(ctx *command.Context, comment *CommentCommand) ([]command.ProjectContext, error) {
 	return b.buildAndEmitStats(
 		"import",
 		func() ([]command.ProjectContext, error) {
@@ -49,6 +49,15 @@ func (b *InstrumentedProjectCommandBuilder) BuildVersionCommands(ctx *command.Co
 	)
 }

+func (b *InstrumentedProjectCommandBuilder) BuildStateRmCommands(ctx *command.Context, comment *CommentCommand) ([]command.ProjectContext, error) {
+	return b.buildAndEmitStats(
+		"state rm",
+		func() ([]command.ProjectContext, error) {
+			return b.ProjectCommandBuilder.BuildStateRmCommands(ctx, comment)
+		},
+	)
+}
+
 func (b *InstrumentedProjectCommandBuilder) buildAndEmitStats(
 	command string,
 	execute func() ([]command.ProjectContext, error),
--- a/server/events/instrumented_project_command_runner.go
+++ b/server/events/instrumented_project_command_runner.go
@@ -12,6 +12,7 @@ type IntrumentedCommandRunner interface {
 	Apply(ctx command.ProjectContext) command.ProjectResult
 	ApprovePolicies(ctx command.ProjectContext) command.ProjectResult
 	Import(ctx command.ProjectContext) command.ProjectResult
+	StateRm(ctx command.ProjectContext) command.ProjectResult
 }

 type InstrumentedProjectCommandRunner struct {
@@ -53,6 +54,10 @@ func (p *InstrumentedProjectCommandRunner) Import(ctx command.ProjectContext) co
 	return RunAndEmitStats("import", ctx, p.projectCommandRunner.Import, p.scope)
 }

+func (p *InstrumentedProjectCommandRunner) StateRm(ctx command.ProjectContext) command.ProjectResult {
+	return RunAndEmitStats("state rm", ctx, p.projectCommandRunner.StateRm, p.scope)
+}
+
 func RunAndEmitStats(commandName string, ctx command.ProjectContext, execute func(ctx command.ProjectContext) command.ProjectResult, scope tally.Scope) command.ProjectResult {
 	// ensures we are differentiating between project level command and overall command
 	scope = ctx.SetProjectScopeTags(scope)
--- a/server/events/markdown_renderer.go
+++ b/server/events/markdown_renderer.go
@@ -34,6 +34,7 @@ var (
 	approvePoliciesCommandTitle = command.ApprovePolicies.TitleString()
 	versionCommandTitle         = command.Version.TitleString()
 	importCommandTitle          = command.Import.TitleString()
+	stateCommandTitle           = command.State.TitleString()
 	// maxUnwrappedLines is the maximum number of lines the Terraform output
 	// can be before we wrap it in an expandable template.
 	maxUnwrappedLines = 12
@@ -60,6 +61,7 @@ type MarkdownRenderer struct {
 // commonData is data that all responses have.
 type commonData struct {
 	Command                  string
+	SubCommand               string
 	Verbose                  bool
 	Log                      string
 	PlansDeleted             bool
@@ -140,10 +142,11 @@ func NewMarkdownRenderer(

 // Render formats the data into a markdown string.
 // nolint: interfacer
-func (m *MarkdownRenderer) Render(res command.Result, cmdName command.Name, log string, verbose bool, vcsHost models.VCSHostType) string {
+func (m *MarkdownRenderer) Render(res command.Result, cmdName command.Name, subCmd, log string, verbose bool, vcsHost models.VCSHostType) string {
 	commandStr := cases.Title(language.English).String(strings.Replace(cmdName.String(), "_", " ", -1))
 	common := commonData{
 		Command:                  commandStr,
+		SubCommand:               subCmd,
 		Verbose:                  verbose,
 		Log:                      log,
 		PlansDeleted:             res.PlansDeleted,
@@ -225,6 +228,13 @@ func (m *MarkdownRenderer) renderProjectResults(results []command.ProjectResult,
 			} else {
 				resultData.Rendered = m.renderTemplateTrimSpace(templates.Lookup("importSuccessUnwrapped"), result.ImportSuccess)
 			}
+		} else if result.StateRmSuccess != nil {
+			result.StateRmSuccess.Output = strings.TrimSpace(result.StateRmSuccess.Output)
+			if m.shouldUseWrappedTmpl(vcsHost, result.StateRmSuccess.Output) {
+				resultData.Rendered = m.renderTemplateTrimSpace(templates.Lookup("stateRmSuccessWrapped"), result.StateRmSuccess)
+			} else {
+				resultData.Rendered = m.renderTemplateTrimSpace(templates.Lookup("stateRmSuccessUnwrapped"), result.StateRmSuccess)
+			}
 		} else {
 			resultData.Rendered = "Found no template. This is a bug!"
 		}
@@ -249,6 +259,13 @@ func (m *MarkdownRenderer) renderProjectResults(results []command.ProjectResult,
 		tmpl = templates.Lookup("singleProjectApply")
 	case len(resultsTmplData) == 1 && common.Command == importCommandTitle:
 		tmpl = templates.Lookup("singleProjectImport")
+	case len(resultsTmplData) == 1 && common.Command == stateCommandTitle:
+		switch common.SubCommand {
+		case "rm":
+			tmpl = templates.Lookup("singleProjectStateRm")
+		default:
+			return fmt.Sprintf("no template matched–this is a bug: command=%s, subcommand=%s", common.Command, common.SubCommand)
+		}
 	case common.Command == planCommandTitle,
 		common.Command == policyCheckCommandTitle:
 		tmpl = templates.Lookup("multiProjectPlan")
@@ -260,8 +277,15 @@ func (m *MarkdownRenderer) renderProjectResults(results []command.ProjectResult,
 		tmpl = templates.Lookup("multiProjectVersion")
 	case common.Command == importCommandTitle:
 		tmpl = templates.Lookup("multiProjectImport")
+	case common.Command == stateCommandTitle:
+		switch common.SubCommand {
+		case "rm":
+			tmpl = templates.Lookup("multiProjectStateRm")
+		default:
+			return fmt.Sprintf("no template matched–this is a bug: command=%s, subcommand=%s", common.Command, common.SubCommand)
+		}
 	default:
-		return "no template matched–this is a bug"
+		return fmt.Sprintf("no template matched–this is a bug: command=%s", common.Command)
 	}
 	return m.renderTemplateTrimSpace(tmpl, resultData{resultsTmplData, common})
 }
--- a/server/events/markdown_renderer_test.go
+++ b/server/events/markdown_renderer_test.go
@@ -64,7 +64,7 @@ func TestRenderErr(t *testing.T) {
 		}
 		for _, verbose := range []bool{true, false} {
 			t.Run(fmt.Sprintf("%s_%t", c.Description, verbose), func(t *testing.T) {
-				s := r.Render(res, c.Command, "log", verbose, models.Github)
+				s := r.Render(res, c.Command, "", "log", verbose, models.Github)
 				if !verbose {
 					Equals(t, strings.TrimSpace(c.Expected), strings.TrimSpace(s))
 				} else {
@@ -109,7 +109,7 @@ func TestRenderFailure(t *testing.T) {
 		}
 		for _, verbose := range []bool{true, false} {
 			t.Run(fmt.Sprintf("%s_%t", c.Description, verbose), func(t *testing.T) {
-				s := r.Render(res, c.Command, "log", verbose, models.Github)
+				s := r.Render(res, c.Command, "", "log", verbose, models.Github)
 				if !verbose {
 					Equals(t, strings.TrimSpace(c.Expected), strings.TrimSpace(s))
 				} else {
@@ -126,7 +126,7 @@ func TestRenderErrAndFailure(t *testing.T) {
 		Error:   errors.New("error"),
 		Failure: "failure",
 	}
-	s := r.Render(res, command.Plan, "", false, models.Github)
+	s := r.Render(res, command.Plan, "", "", false, models.Github)
 	Equals(t, "**Plan Error**\n```\nerror\n```", s)
 }

@@ -134,6 +134,7 @@ func TestRenderProjectResults(t *testing.T) {
 	cases := []struct {
 		Description    string
 		Command        command.Name
+		SubCommand     string
 		ProjectResults []command.ProjectResult
 		VCSHost        models.VCSHostType
 		Expected       string
@@ -141,6 +142,7 @@ func TestRenderProjectResults(t *testing.T) {
 		{
 			"no projects",
 			command.Plan,
+			"",
 			[]command.ProjectResult{},
 			models.Github,
 			"Ran Plan for 0 projects:\n\n\n",
@@ -148,6 +150,7 @@ func TestRenderProjectResults(t *testing.T) {
 		{
 			"single successful plan",
 			command.Plan,
+			"",
 			[]command.ProjectResult{
 				{
 					PlanSuccess: &models.PlanSuccess{
@@ -183,6 +186,7 @@ $$$
 		{
 			"single successful plan with main ahead",
 			command.Plan,
+			"",
 			[]command.ProjectResult{
 				{
 					PlanSuccess: &models.PlanSuccess{
@@ -221,6 +225,7 @@ $$$
 		{
 			"single successful plan with project name",
 			command.Plan,
+			"",
 			[]command.ProjectResult{
 				{
 					PlanSuccess: &models.PlanSuccess{
@@ -257,6 +262,7 @@ $$$
 		{
 			"single successful policy check with project name",
 			command.PolicyCheck,
+			"",
 			[]command.ProjectResult{
 				{
 					PolicyCheckSuccess: &models.PolicyCheckSuccess{
@@ -306,6 +312,7 @@ $$$
 		{
 			"single successful import",
 			command.Import,
+			"",
 			[]command.ProjectResult{
 				{
 					ImportSuccess: &models.ImportSuccess{
@@ -324,12 +331,43 @@ $$$diff
 import-output
 $$$

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * $atlantis plan -d path -w workspace$`,
+		},
+		{
+			"single successful state rm",
+			command.State,
+			"rm",
+			[]command.ProjectResult{
+				{
+					StateRmSuccess: &models.StateRmSuccess{
+						Output:    "state-rm-output",
+						RePlanCmd: "atlantis plan -d path -w workspace",
+					},
+					Workspace:   "workspace",
+					RepoRelDir:  "path",
+					ProjectName: "projectname",
+				},
+			},
+			models.Github,
+			`Ran State $rm$ for project: $projectname$ dir: $path$ workspace: $workspace$
+
+$$$diff
+state-rm-output
+$$$
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * $atlantis plan -d path -w workspace$
+`,
 		},
 		{
 			"single successful apply",
 			command.Apply,
+			"",
 			[]command.ProjectResult{
 				{
 					ApplySuccess: "success",
@@ -347,6 +385,7 @@ $$$`,
 		{
 			"single successful apply with project name",
 			command.Apply,
+			"",
 			[]command.ProjectResult{
 				{
 					ApplySuccess: "success",
@@ -365,6 +404,7 @@ $$$`,
 		{
 			"multiple successful plans",
 			command.Plan,
+			"",
 			[]command.ProjectResult{
 				{
 					Workspace:  "workspace",
@@ -427,6 +467,7 @@ $$$
 		{
 			"multiple successful policy checks",
 			command.PolicyCheck,
+			"",
 			[]command.ProjectResult{
 				{
 					Workspace:  "workspace",
@@ -489,6 +530,7 @@ $$$
 		{
 			"multiple successful applies",
 			command.Apply,
+			"",
 			[]command.ProjectResult{
 				{
 					RepoRelDir:   "path",
@@ -525,6 +567,7 @@ $$$
 		{
 			"single errored plan",
 			command.Plan,
+			"",
 			[]command.ProjectResult{
 				{
 					Error:      errors.New("error"),
@@ -543,6 +586,7 @@ $$$`,
 		{
 			"single failed plan",
 			command.Plan,
+			"",
 			[]command.ProjectResult{
 				{
 					RepoRelDir: "path",
@@ -558,6 +602,7 @@ $$$`,
 		{
 			"successful, failed, and errored plan",
 			command.Plan,
+			"",
 			[]command.ProjectResult{
 				{
 					Workspace:  "workspace",
@@ -620,6 +665,7 @@ $$$
 		{
 			"successful, failed, and errored policy check",
 			command.PolicyCheck,
+			"",
 			[]command.ProjectResult{
 				{
 					Workspace:  "workspace",
@@ -685,6 +731,7 @@ $$$
 		{
 			"successful, failed, and errored apply",
 			command.Apply,
+			"",
 			[]command.ProjectResult{
 				{
 					Workspace:    "workspace",
@@ -731,6 +778,7 @@ $$$
 		{
 			"successful, failed, and errored apply",
 			command.Apply,
+			"",
 			[]command.ProjectResult{
 				{
 					Workspace:    "workspace",
@@ -784,7 +832,7 @@ $$$
 			}
 			for _, verbose := range []bool{true, false} {
 				t.Run(c.Description, func(t *testing.T) {
-					s := r.Render(res, c.Command, "log", verbose, c.VCSHost)
+					s := r.Render(res, c.Command, c.SubCommand, "log", verbose, c.VCSHost)
 					expWithBackticks := strings.Replace(c.Expected, "$", "`", -1)
 					if !verbose {
 						Equals(t, strings.TrimSpace(expWithBackticks), strings.TrimSpace(s))
@@ -939,7 +987,7 @@ $$$
 			}
 			for _, verbose := range []bool{true, false} {
 				t.Run(c.Description, func(t *testing.T) {
-					s := r.Render(res, c.Command, "log", verbose, c.VCSHost)
+					s := r.Render(res, c.Command, "", "log", verbose, c.VCSHost)
 					expWithBackticks := strings.Replace(c.Expected, "$", "`", -1)
 					if !verbose {
 						Equals(t, strings.TrimSpace(expWithBackticks), strings.TrimSpace(s))
@@ -1087,7 +1135,7 @@ $$$
 			}
 			for _, verbose := range []bool{true, false} {
 				t.Run(c.Description, func(t *testing.T) {
-					s := r.Render(res, c.Command, "log", verbose, c.VCSHost)
+					s := r.Render(res, c.Command, "", "log", verbose, c.VCSHost)
 					expWithBackticks := strings.Replace(c.Expected, "$", "`", -1)
 					if !verbose {
 						Equals(t, strings.TrimSpace(expWithBackticks), strings.TrimSpace(s))
@@ -1132,7 +1180,7 @@ func TestRenderCustomPolicyCheckTemplate_DisableApplyAll(t *testing.T) {
 				},
 			},
 		},
-	}, command.PolicyCheck, "log", false, models.Github)
+	}, command.PolicyCheck, "", "log", false, models.Github)
 	exp := "Ran Policy Check for dir: `path` workspace: `workspace`\n\nsomecustometext"
 	Equals(t, exp, rendered)
 }
@@ -1158,7 +1206,7 @@ func TestRenderProjectResults_DisableFolding(t *testing.T) {
 				Error:      errors.New(strings.Repeat("line\n", 13)),
 			},
 		},
-	}, command.Plan, "log", false, models.Github)
+	}, command.Plan, "", "log", false, models.Github)
 	Equals(t, false, strings.Contains(rendered, "\n<details>"))
 }

@@ -1249,7 +1297,7 @@ func TestRenderProjectResults_WrappedErr(t *testing.T) {
 							Error:      errors.New(c.Output),
 						},
 					},
-				}, command.Plan, "log", false, c.VCSHost)
+				}, command.Plan, "", "log", false, c.VCSHost)
 				var exp string
 				if c.ShouldWrap {
 					exp = `Ran Plan for dir: $.$ workspace: $default$
@@ -1377,7 +1425,7 @@ func TestRenderProjectResults_WrapSingleProject(t *testing.T) {
 					}
 					rendered := mr.Render(command.Result{
 						ProjectResults: []command.ProjectResult{pr},
-					}, cmd, "log", false, c.VCSHost)
+					}, cmd, "", "log", false, c.VCSHost)

 					// Check result.
 					var exp string
@@ -1476,7 +1524,7 @@ func TestRenderProjectResults_MultiProjectApplyWrapped(t *testing.T) {
 				ApplySuccess: tfOut,
 			},
 		},
-	}, command.Apply, "log", false, models.Github)
+	}, command.Apply, "", "log", false, models.Github)
 	exp := `Ran Apply for 2 projects:

 1. dir: $.$ workspace: $staging$
@@ -1541,7 +1589,7 @@ func TestRenderProjectResults_MultiProjectPlanWrapped(t *testing.T) {
 				},
 			},
 		},
-	}, command.Plan, "log", false, models.Github)
+	}, command.Plan, "", "log", false, models.Github)
 	exp := `Ran Plan for 2 projects:

 1. dir: $.$ workspace: $staging$
@@ -1692,7 +1740,7 @@ This plan was not saved because one or more projects failed and automerge requir
 				"",         // MarkdownTemplateOverridesDir
 				"atlantis", // executableName
 			)
-			rendered := mr.Render(c.cr, command.Plan, "log", false, models.Github)
+			rendered := mr.Render(c.cr, command.Plan, "", "log", false, models.Github)
 			expWithBackticks := strings.Replace(c.exp, "$", "`", -1)
 			Equals(t, expWithBackticks, rendered)
 		})
@@ -2157,7 +2205,7 @@ $$$
 			}
 			for _, verbose := range []bool{true, false} {
 				t.Run(c.Description, func(t *testing.T) {
-					s := r.Render(res, c.Command, "log", verbose, c.VCSHost)
+					s := r.Render(res, c.Command, "", "log", verbose, c.VCSHost)
 					expWithBackticks := strings.Replace(c.Expected, "$", "`", -1)
 					if !verbose {
 						Equals(t, strings.TrimSpace(expWithBackticks), strings.TrimSpace(s))
@@ -2594,7 +2642,7 @@ func TestRenderProjectResultsWithEnableDiffMarkdownFormat(t *testing.T) {
 			}
 			for _, verbose := range []bool{true, false} {
 				t.Run(c.Description, func(t *testing.T) {
-					s := r.Render(res, c.Command, "log", verbose, c.VCSHost)
+					s := r.Render(res, c.Command, "", "log", verbose, c.VCSHost)
 					expWithBackticks := strings.Replace(c.Expected, "$", "`", -1)
 					if !verbose {
 						Equals(t, strings.TrimSpace(expWithBackticks), strings.TrimSpace(s))
@@ -2632,7 +2680,7 @@ func BenchmarkRenderProjectResultsWithEnableDiffMarkdownFormat(b *testing.B) {
 				b.Run(fmt.Sprintf("verbose %t", verbose), func(b *testing.B) {
 					b.ReportAllocs()
 					for i := 0; i < b.N; i++ {
-						render = r.Render(res, c.Command, "log", verbose, c.VCSHost)
+						render = r.Render(res, c.Command, "", "log", verbose, c.VCSHost)
 					}
 					Render = render
 				})
--- a/server/events/mocks/mock_project_command_builder.go
+++ b/server/events/mocks/mock_project_command_builder.go
@@ -121,6 +121,25 @@ func (mock *MockProjectCommandBuilder) BuildPlanCommands(_param0 *command.Contex
 	return ret0, ret1
 }

+func (mock *MockProjectCommandBuilder) BuildStateRmCommands(_param0 *command.Context, _param1 *events.CommentCommand) ([]command.ProjectContext, error) {
+	if mock == nil {
+		panic("mock must not be nil. Use myMock := NewMockProjectCommandBuilder().")
+	}
+	params := []pegomock.Param{_param0, _param1}
+	result := pegomock.GetGenericMockFrom(mock).Invoke("BuildStateRmCommands", params, []reflect.Type{reflect.TypeOf((*[]command.ProjectContext)(nil)).Elem(), reflect.TypeOf((*error)(nil)).Elem()})
+	var ret0 []command.ProjectContext
+	var ret1 error
+	if len(result) != 0 {
+		if result[0] != nil {
+			ret0 = result[0].([]command.ProjectContext)
+		}
+		if result[1] != nil {
+			ret1 = result[1].(error)
+		}
+	}
+	return ret0, ret1
+}
+
 func (mock *MockProjectCommandBuilder) BuildVersionCommands(_param0 *command.Context, _param1 *events.CommentCommand) ([]command.ProjectContext, error) {
 	if mock == nil {
 		panic("mock must not be nil. Use myMock := NewMockProjectCommandBuilder().")
@@ -328,6 +347,37 @@ func (c *MockProjectCommandBuilder_BuildPlanCommands_OngoingVerification) GetAll
 	return
 }

+func (verifier *VerifierMockProjectCommandBuilder) BuildStateRmCommands(_param0 *command.Context, _param1 *events.CommentCommand) *MockProjectCommandBuilder_BuildStateRmCommands_OngoingVerification {
+	params := []pegomock.Param{_param0, _param1}
+	methodInvocations := pegomock.GetGenericMockFrom(verifier.mock).Verify(verifier.inOrderContext, verifier.invocationCountMatcher, "BuildStateRmCommands", params, verifier.timeout)
+	return &MockProjectCommandBuilder_BuildStateRmCommands_OngoingVerification{mock: verifier.mock, methodInvocations: methodInvocations}
+}
+
+type MockProjectCommandBuilder_BuildStateRmCommands_OngoingVerification struct {
+	mock              *MockProjectCommandBuilder
+	methodInvocations []pegomock.MethodInvocation
+}
+
+func (c *MockProjectCommandBuilder_BuildStateRmCommands_OngoingVerification) GetCapturedArguments() (*command.Context, *events.CommentCommand) {
+	_param0, _param1 := c.GetAllCapturedArguments()
+	return _param0[len(_param0)-1], _param1[len(_param1)-1]
+}
+
+func (c *MockProjectCommandBuilder_BuildStateRmCommands_OngoingVerification) GetAllCapturedArguments() (_param0 []*command.Context, _param1 []*events.CommentCommand) {
+	params := pegomock.GetGenericMockFrom(c.mock).GetInvocationParams(c.methodInvocations)
+	if len(params) > 0 {
+		_param0 = make([]*command.Context, len(c.methodInvocations))
+		for u, param := range params[0] {
+			_param0[u] = param.(*command.Context)
+		}
+		_param1 = make([]*events.CommentCommand, len(c.methodInvocations))
+		for u, param := range params[1] {
+			_param1[u] = param.(*events.CommentCommand)
+		}
+	}
+	return
+}
+
 func (verifier *VerifierMockProjectCommandBuilder) BuildVersionCommands(_param0 *command.Context, _param1 *events.CommentCommand) *MockProjectCommandBuilder_BuildVersionCommands_OngoingVerification {
 	params := []pegomock.Param{_param0, _param1}
 	methodInvocations := pegomock.GetGenericMockFrom(verifier.mock).Verify(verifier.inOrderContext, verifier.invocationCountMatcher, "BuildVersionCommands", params, verifier.timeout)
--- a/server/events/mocks/mock_project_command_runner.go
+++ b/server/events/mocks/mock_project_command_runner.go
@@ -100,6 +100,21 @@ func (mock *MockProjectCommandRunner) PolicyCheck(_param0 command.ProjectContext
 	return ret0
 }

+func (mock *MockProjectCommandRunner) StateRm(_param0 command.ProjectContext) command.ProjectResult {
+	if mock == nil {
+		panic("mock must not be nil. Use myMock := NewMockProjectCommandRunner().")
+	}
+	params := []pegomock.Param{_param0}
+	result := pegomock.GetGenericMockFrom(mock).Invoke("StateRm", params, []reflect.Type{reflect.TypeOf((*command.ProjectResult)(nil)).Elem()})
+	var ret0 command.ProjectResult
+	if len(result) != 0 {
+		if result[0] != nil {
+			ret0 = result[0].(command.ProjectResult)
+		}
+	}
+	return ret0
+}
+
 func (mock *MockProjectCommandRunner) Version(_param0 command.ProjectContext) command.ProjectResult {
 	if mock == nil {
 		panic("mock must not be nil. Use myMock := NewMockProjectCommandRunner().")
@@ -287,6 +302,33 @@ func (c *MockProjectCommandRunner_PolicyCheck_OngoingVerification) GetAllCapture
 	return
 }

+func (verifier *VerifierMockProjectCommandRunner) StateRm(_param0 command.ProjectContext) *MockProjectCommandRunner_StateRm_OngoingVerification {
+	params := []pegomock.Param{_param0}
+	methodInvocations := pegomock.GetGenericMockFrom(verifier.mock).Verify(verifier.inOrderContext, verifier.invocationCountMatcher, "StateRm", params, verifier.timeout)
+	return &MockProjectCommandRunner_StateRm_OngoingVerification{mock: verifier.mock, methodInvocations: methodInvocations}
+}
+
+type MockProjectCommandRunner_StateRm_OngoingVerification struct {
+	mock              *MockProjectCommandRunner
+	methodInvocations []pegomock.MethodInvocation
+}
+
+func (c *MockProjectCommandRunner_StateRm_OngoingVerification) GetCapturedArguments() command.ProjectContext {
+	_param0 := c.GetAllCapturedArguments()
+	return _param0[len(_param0)-1]
+}
+
+func (c *MockProjectCommandRunner_StateRm_OngoingVerification) GetAllCapturedArguments() (_param0 []command.ProjectContext) {
+	params := pegomock.GetGenericMockFrom(c.mock).GetInvocationParams(c.methodInvocations)
+	if len(params) > 0 {
+		_param0 = make([]command.ProjectContext, len(c.methodInvocations))
+		for u, param := range params[0] {
+			_param0[u] = param.(command.ProjectContext)
+		}
+	}
+	return
+}
+
 func (verifier *VerifierMockProjectCommandRunner) Version(_param0 command.ProjectContext) *MockProjectCommandRunner_Version_OngoingVerification {
 	params := []pegomock.Param{_param0}
 	methodInvocations := pegomock.GetGenericMockFrom(verifier.mock).Verify(verifier.inOrderContext, verifier.invocationCountMatcher, "Version", params, verifier.timeout)
--- a/server/events/models/models.go
+++ b/server/events/models/models.go
@@ -430,6 +430,14 @@ type ImportSuccess struct {
 	RePlanCmd string
 }

+// StateRmSuccess is the result of a successful state rm run.
+type StateRmSuccess struct {
+	// Output is the output from terraform state rm
+	Output string
+	// RePlanCmd is the command that users should run to re-plan this project.
+	RePlanCmd string
+}
+
 // Summary extracts one line summary of policy check.
 func (p *PolicyCheckSuccess) Summary() string {
 	note := ""
--- a/server/events/project_command_builder.go
+++ b/server/events/project_command_builder.go
@@ -165,6 +165,13 @@ type ProjectImportCommandBuilder interface {
 	BuildImportCommands(ctx *command.Context, comment *CommentCommand) ([]command.ProjectContext, error)
 }

+type ProjectStateCommandBuilder interface {
+	// BuildStateRmCommands builds project state rm commands for this ctx and comment. If
+	// comment doesn't specify one project then there may be multiple commands
+	// to be run.
+	BuildStateRmCommands(ctx *command.Context, comment *CommentCommand) ([]command.ProjectContext, error)
+}
+
 //go:generate pegomock generate -m --package mocks -o mocks/mock_project_command_builder.go ProjectCommandBuilder

 // ProjectCommandBuilder builds commands that run on individual projects.
@@ -174,6 +181,7 @@ type ProjectCommandBuilder interface {
 	ProjectApprovePoliciesCommandBuilder
 	ProjectVersionCommandBuilder
 	ProjectImportCommandBuilder
+	ProjectStateCommandBuilder
 }

 // DefaultProjectCommandBuilder implements ProjectCommandBuilder.
@@ -199,7 +207,7 @@ type DefaultProjectCommandBuilder struct {

 // See ProjectCommandBuilder.BuildAutoplanCommands.
 func (p *DefaultProjectCommandBuilder) BuildAutoplanCommands(ctx *command.Context) ([]command.ProjectContext, error) {
-	projCtxs, err := p.buildAllCommandsByCfg(ctx, command.Plan, nil, false)
+	projCtxs, err := p.buildAllCommandsByCfg(ctx, command.Plan, "", nil, false)
 	if err != nil {
 		return nil, err
 	}
@@ -217,7 +225,7 @@ func (p *DefaultProjectCommandBuilder) BuildAutoplanCommands(ctx *command.Contex
 // See ProjectCommandBuilder.BuildPlanCommands.
 func (p *DefaultProjectCommandBuilder) BuildPlanCommands(ctx *command.Context, cmd *CommentCommand) ([]command.ProjectContext, error) {
 	if !cmd.IsForSpecificProject() {
-		return p.buildAllCommandsByCfg(ctx, cmd.CommandName(), cmd.Flags, cmd.Verbose)
+		return p.buildAllCommandsByCfg(ctx, cmd.CommandName(), cmd.SubName, cmd.Flags, cmd.Verbose)
 	}
 	pcc, err := p.buildProjectPlanCommand(ctx, cmd)
 	return pcc, err
@@ -228,7 +236,7 @@ func (p *DefaultProjectCommandBuilder) BuildApplyCommands(ctx *command.Context,
 	if !cmd.IsForSpecificProject() {
 		return p.buildAllProjectCommandsByPlan(ctx, cmd)
 	}
-	pac, err := p.buildProjectApplyCommand(ctx, cmd)
+	pac, err := p.buildProjectCommand(ctx, cmd)
 	return pac, err
 }

@@ -240,21 +248,29 @@ func (p *DefaultProjectCommandBuilder) BuildVersionCommands(ctx *command.Context
 	if !cmd.IsForSpecificProject() {
 		return p.buildAllProjectCommandsByPlan(ctx, cmd)
 	}
-	pac, err := p.buildProjectVersionCommand(ctx, cmd)
+	pac, err := p.buildProjectCommand(ctx, cmd)
 	return pac, err
 }

 func (p *DefaultProjectCommandBuilder) BuildImportCommands(ctx *command.Context, cmd *CommentCommand) ([]command.ProjectContext, error) {
 	if !cmd.IsForSpecificProject() {
 		// import discard a plan file, so use buildAllCommandsByCfg instead buildAllProjectCommandsByPlan.
-		return p.buildAllCommandsByCfg(ctx, cmd.CommandName(), cmd.Flags, cmd.Verbose)
+		return p.buildAllCommandsByCfg(ctx, cmd.CommandName(), cmd.SubName, cmd.Flags, cmd.Verbose)
 	}
-	return p.buildProjectImportCommand(ctx, cmd)
+	return p.buildProjectCommand(ctx, cmd)
+}
+
+func (p *DefaultProjectCommandBuilder) BuildStateRmCommands(ctx *command.Context, cmd *CommentCommand) ([]command.ProjectContext, error) {
+	if !cmd.IsForSpecificProject() {
+		// state rm discard a plan file, so use buildAllCommandsByCfg instead buildAllProjectCommandsByPlan.
+		return p.buildAllCommandsByCfg(ctx, cmd.CommandName(), cmd.SubName, cmd.Flags, cmd.Verbose)
+	}
+	return p.buildProjectCommand(ctx, cmd)
 }

 // buildAllCommandsByCfg builds init contexts for all projects we determine were
 // modified in this ctx.
-func (p *DefaultProjectCommandBuilder) buildAllCommandsByCfg(ctx *command.Context, cmdName command.Name, commentFlags []string, verbose bool) ([]command.ProjectContext, error) {
+func (p *DefaultProjectCommandBuilder) buildAllCommandsByCfg(ctx *command.Context, cmdName command.Name, subCmdName string, commentFlags []string, verbose bool) ([]command.ProjectContext, error) {
 	// We'll need the list of modified files.
 	modifiedFiles, err := p.VCSClient.GetModifiedFiles(ctx.Pull.BaseRepo, ctx.Pull)
 	if err != nil {
@@ -345,6 +361,7 @@ func (p *DefaultProjectCommandBuilder) buildAllCommandsByCfg(ctx *command.Contex
 				p.ProjectCommandContextBuilder.BuildProjectContext(
 					ctx,
 					cmdName,
+					subCmdName,
 					mergedCfg,
 					commentFlags,
 					repoDir,
@@ -391,6 +408,7 @@ func (p *DefaultProjectCommandBuilder) buildAllCommandsByCfg(ctx *command.Contex
 				p.ProjectCommandContextBuilder.BuildProjectContext(
 					ctx,
 					cmdName,
+					subCmdName,
 					pCfg,
 					commentFlags,
 					repoDir,
@@ -498,6 +516,7 @@ func (p *DefaultProjectCommandBuilder) buildProjectPlanCommand(ctx *command.Cont
 	return p.buildProjectCommandCtx(
 		ctx,
 		command.Plan,
+		"",
 		cmd.ProjectName,
 		cmd.Flags,
 		defaultRepoDir,
@@ -590,7 +609,7 @@ func (p *DefaultProjectCommandBuilder) buildAllProjectCommandsByPlan(ctx *comman

 	var cmds []command.ProjectContext
 	for _, plan := range plans {
-		commentCmds, err := p.buildProjectCommandCtx(ctx, commentCmd.CommandName(), plan.ProjectName, commentCmd.Flags, defaultRepoDir, plan.RepoRelDir, plan.Workspace, commentCmd.Verbose)
+		commentCmds, err := p.buildProjectCommandCtx(ctx, commentCmd.CommandName(), commentCmd.SubName, plan.ProjectName, commentCmd.Flags, defaultRepoDir, plan.RepoRelDir, plan.Workspace, commentCmd.Verbose)
 		if err != nil {
 			return nil, errors.Wrapf(err, "building command for dir %q", plan.RepoRelDir)
 		}
@@ -604,9 +623,9 @@ func (p *DefaultProjectCommandBuilder) buildAllProjectCommandsByPlan(ctx *comman
 	return cmds, nil
 }

-// buildProjectApplyCommand builds an apply command for the single project
-// identified by cmd.
-func (p *DefaultProjectCommandBuilder) buildProjectApplyCommand(ctx *command.Context, cmd *CommentCommand) ([]command.ProjectContext, error) {
+// buildProjectCommand builds an command for the single project
+// identified by cmd except plan.
+func (p *DefaultProjectCommandBuilder) buildProjectCommand(ctx *command.Context, cmd *CommentCommand) ([]command.ProjectContext, error) {
 	workspace := DefaultWorkspace
 	if cmd.Workspace != "" {
 		workspace = cmd.Workspace
@@ -635,89 +654,8 @@ func (p *DefaultProjectCommandBuilder) buildProjectApplyCommand(ctx *command.Con

 	return p.buildProjectCommandCtx(
 		ctx,
-		command.Apply,
-		cmd.ProjectName,
-		cmd.Flags,
-		repoDir,
-		repoRelDir,
-		workspace,
-		cmd.Verbose,
-	)
-}
-
-// buildProjectVersionCommand builds a version command for the single project
-// identified by cmd.
-func (p *DefaultProjectCommandBuilder) buildProjectVersionCommand(ctx *command.Context, cmd *CommentCommand) ([]command.ProjectContext, error) {
-	workspace := DefaultWorkspace
-	if cmd.Workspace != "" {
-		workspace = cmd.Workspace
-	}
-
-	var projCtx []command.ProjectContext
-	unlockFn, err := p.WorkingDirLocker.TryLock(ctx.Pull.BaseRepo.FullName, ctx.Pull.Num, workspace, DefaultRepoRelDir)
-	if err != nil {
-		return projCtx, err
-	}
-	defer unlockFn()
-
-	// use the default repository workspace because it is the only one guaranteed to have an atlantis.yaml,
-	// other workspaces will not have the file if they are using pre_workflow_hooks to generate it dynamically
-	repoDir, err := p.WorkingDir.GetWorkingDir(ctx.Pull.BaseRepo, ctx.Pull, DefaultWorkspace)
-	if os.IsNotExist(errors.Cause(err)) {
-		return projCtx, errors.New("no working directory found–did you run plan?")
-	} else if err != nil {
-		return projCtx, err
-	}
-
-	repoRelDir := DefaultRepoRelDir
-	if cmd.RepoRelDir != "" {
-		repoRelDir = cmd.RepoRelDir
-	}
-
-	return p.buildProjectCommandCtx(
-		ctx,
-		command.Version,
-		cmd.ProjectName,
-		cmd.Flags,
-		repoDir,
-		repoRelDir,
-		workspace,
-		cmd.Verbose,
-	)
-}
-
-// buildProjectImportCommand builds a import command for the single project
-// identified by cmd.
-func (p *DefaultProjectCommandBuilder) buildProjectImportCommand(ctx *command.Context, cmd *CommentCommand) ([]command.ProjectContext, error) {
-	workspace := DefaultWorkspace
-	if cmd.Workspace != "" {
-		workspace = cmd.Workspace
-	}
-
-	var projCtx []command.ProjectContext
-	unlockFn, err := p.WorkingDirLocker.TryLock(ctx.Pull.BaseRepo.FullName, ctx.Pull.Num, workspace, DefaultRepoRelDir)
-	if err != nil {
-		return projCtx, err
-	}
-	defer unlockFn()
-
-	// use the default repository workspace because it is the only one guaranteed to have an atlantis.yaml,
-	// other workspaces will not have the file if they are using pre_workflow_hooks to generate it dynamically
-	repoDir, err := p.WorkingDir.GetWorkingDir(ctx.Pull.BaseRepo, ctx.Pull, DefaultWorkspace)
-	if os.IsNotExist(errors.Cause(err)) {
-		return projCtx, errors.New("no working directory found–did you run plan?")
-	} else if err != nil {
-		return projCtx, err
-	}
-
-	repoRelDir := DefaultRepoRelDir
-	if cmd.RepoRelDir != "" {
-		repoRelDir = cmd.RepoRelDir
-	}
-
-	return p.buildProjectCommandCtx(
-		ctx,
-		command.Import,
+		cmd.Name,
+		cmd.SubName,
 		cmd.ProjectName,
 		cmd.Flags,
 		repoDir,
@@ -731,6 +669,7 @@ func (p *DefaultProjectCommandBuilder) buildProjectImportCommand(ctx *command.Co
 // by the parameters.
 func (p *DefaultProjectCommandBuilder) buildProjectCommandCtx(ctx *command.Context,
 	cmd command.Name,
+	subCmd string,
 	projectName string,
 	commentFlags []string,
 	repoDir string,
@@ -767,6 +706,7 @@ func (p *DefaultProjectCommandBuilder) buildProjectCommandCtx(ctx *command.Conte
 				p.ProjectCommandContextBuilder.BuildProjectContext(
 					ctx,
 					cmd,
+					subCmd,
 					projCfg,
 					commentFlags,
 					repoDir,
@@ -783,6 +723,7 @@ func (p *DefaultProjectCommandBuilder) buildProjectCommandCtx(ctx *command.Conte
 			p.ProjectCommandContextBuilder.BuildProjectContext(
 				ctx,
 				cmd,
+				subCmd,
 				projCfg,
 				commentFlags,
 				repoDir,
--- a/server/events/project_command_builder_internal_test.go
+++ b/server/events/project_command_builder_internal_test.go
@@ -668,7 +668,7 @@ projects:
 						PullRequestStatus: models.PullReqStatus{
 							Mergeable: true,
 						},
-					}, cmd, "", []string{"flag"}, tmp, "project1", "myworkspace", true)
+					}, cmd, "", "", []string{"flag"}, tmp, "project1", "myworkspace", true)

 					if c.expErr != "" {
 						ErrEquals(t, c.expErr, err)
@@ -876,7 +876,7 @@ projects:
 						PullRequestStatus: models.PullReqStatus{
 							Mergeable: true,
 						},
-					}, cmd, "myproject_[1-2]", []string{"flag"}, tmp, "project1", "myworkspace", true)
+					}, cmd, "", "myproject_[1-2]", []string{"flag"}, tmp, "project1", "myworkspace", true)

 					if c.expErr != "" {
 						ErrEquals(t, c.expErr, err)
@@ -1113,7 +1113,7 @@ workflows:
 					PullRequestStatus: models.PullReqStatus{
 						Mergeable: true,
 					},
-				}, command.Plan, "", []string{"flag"}, tmp, "project1", "myworkspace", true)
+				}, command.Plan, "", "", []string{"flag"}, tmp, "project1", "myworkspace", true)

 				if c.expErr != "" {
 					ErrEquals(t, c.expErr, err)
--- a/server/events/project_command_context_builder.go
+++ b/server/events/project_command_context_builder.go
@@ -34,6 +34,7 @@ type ProjectCommandContextBuilder interface {
 	BuildProjectContext(
 		ctx *command.Context,
 		cmdName command.Name,
+		subCmdName string,
 		prjCfg valid.MergedProjectCfg,
 		commentFlags []string,
 		repoDir string,
@@ -53,6 +54,7 @@ type CommandScopedStatsProjectCommandContextBuilder struct {
 func (cb *CommandScopedStatsProjectCommandContextBuilder) BuildProjectContext(
 	ctx *command.Context,
 	cmdName command.Name,
+	subCmdName string,
 	prjCfg valid.MergedProjectCfg,
 	commentFlags []string,
 	repoDir string,
@@ -61,7 +63,7 @@ func (cb *CommandScopedStatsProjectCommandContextBuilder) BuildProjectContext(
 	cb.ProjectCounter.Inc(1)

 	cmds := cb.ProjectCommandContextBuilder.BuildProjectContext(
-		ctx, cmdName, prjCfg, commentFlags, repoDir, automerge, parallelApply, parallelPlan, verbose, terraformClient,
+		ctx, cmdName, subCmdName, prjCfg, commentFlags, repoDir, automerge, parallelApply, parallelPlan, verbose, terraformClient,
 	)

 	projectCmds = []command.ProjectContext{}
@@ -85,6 +87,7 @@ type DefaultProjectCommandContextBuilder struct {
 func (cb *DefaultProjectCommandContextBuilder) BuildProjectContext(
 	ctx *command.Context,
 	cmdName command.Name,
+	subName string,
 	prjCfg valid.MergedProjectCfg,
 	commentFlags []string,
 	repoDir string,
@@ -105,6 +108,15 @@ func (cb *DefaultProjectCommandContextBuilder) BuildProjectContext(
 		}}
 	case command.Import:
 		steps = prjCfg.Workflow.Import.Steps
+	case command.State:
+		switch subName {
+		case "rm":
+			steps = prjCfg.Workflow.StateRm.Steps
+		default:
+			// comment_parser prevent invalid subcommand, so not need to handle this.
+			// if comes here, state_command_runner will respond on PR, so it's enough to do log only.
+			ctx.Log.Err("unknown state subcommand: %s", subName)
+		}
 	}

 	// If TerraformVersion not defined in config file look for a
@@ -143,6 +155,7 @@ type PolicyCheckProjectCommandContextBuilder struct {
 func (cb *PolicyCheckProjectCommandContextBuilder) BuildProjectContext(
 	ctx *command.Context,
 	cmdName command.Name,
+	subCmdName string,
 	prjCfg valid.MergedProjectCfg,
 	commentFlags []string,
 	repoDir string,
@@ -159,6 +172,7 @@ func (cb *PolicyCheckProjectCommandContextBuilder) BuildProjectContext(
 	projectCmds = cb.ProjectCommandContextBuilder.BuildProjectContext(
 		ctx,
 		cmdName,
+		subCmdName,
 		prjCfg,
 		commentFlags,
 		repoDir,
--- a/server/events/project_command_context_builder_test.go
+++ b/server/events/project_command_context_builder_test.go
@@ -62,7 +62,7 @@ func TestProjectCommandContextBuilder_PullStatus(t *testing.T) {
 			},
 		}

-		result := subject.BuildProjectContext(commandCtx, command.Plan, projCfg, []string{}, "some/dir", false, false, false, false, terraformClient)
+		result := subject.BuildProjectContext(commandCtx, command.Plan, "", projCfg, []string{}, "some/dir", false, false, false, false, terraformClient)
 		assert.Equal(t, models.ErroredPolicyCheckStatus, result[0].ProjectPlanStatus)
 	})

@@ -81,7 +81,7 @@ func TestProjectCommandContextBuilder_PullStatus(t *testing.T) {
 			},
 		}

-		result := subject.BuildProjectContext(commandCtx, command.Plan, projCfg, []string{}, "some/dir", false, false, false, false, terraformClient)
+		result := subject.BuildProjectContext(commandCtx, command.Plan, "", projCfg, []string{}, "some/dir", false, false, false, false, terraformClient)

 		assert.Equal(t, models.ErroredPolicyCheckStatus, result[0].ProjectPlanStatus)
 	})
@@ -101,7 +101,7 @@ func TestProjectCommandContextBuilder_PullStatus(t *testing.T) {
 			},
 		}

-		result := subject.BuildProjectContext(commandCtx, command.Plan, projCfg, []string{}, "some/dir", false, true, false, false, terraformClient)
+		result := subject.BuildProjectContext(commandCtx, command.Plan, "", projCfg, []string{}, "some/dir", false, true, false, false, terraformClient)

 		assert.True(t, result[0].ParallelApplyEnabled)
 		assert.False(t, result[0].ParallelPlanEnabled)
--- a/server/events/project_command_runner.go
+++ b/server/events/project_command_runner.go
@@ -118,6 +118,11 @@ type ProjectImportCommandRunner interface {
 	Import(ctx command.ProjectContext) command.ProjectResult
 }

+type ProjectStateCommandRunner interface {
+	// StateRm runs terraform state rm for the project described by ctx.
+	StateRm(ctx command.ProjectContext) command.ProjectResult
+}
+
 // ProjectCommandRunner runs project commands. A project command is a command
 // for a specific TF project.
 type ProjectCommandRunner interface {
@@ -127,6 +132,7 @@ type ProjectCommandRunner interface {
 	ProjectApprovePoliciesCommandRunner
 	ProjectVersionCommandRunner
 	ProjectImportCommandRunner
+	ProjectStateCommandRunner
 }

 //go:generate pegomock generate -m --package mocks -o mocks/mock_job_url_setter.go JobURLSetter
@@ -200,6 +206,7 @@ type DefaultProjectCommandRunner struct {
 	PolicyCheckStepRunner     StepRunner
 	VersionStepRunner         StepRunner
 	ImportStepRunner          StepRunner
+	StateRmStepRunner         StepRunner
 	RunStepRunner             CustomStepRunner
 	EnvStepRunner             EnvStepRunner
 	MultiEnvStepRunner        MultiEnvStepRunner
@@ -292,6 +299,21 @@ func (p *DefaultProjectCommandRunner) Import(ctx command.ProjectContext) command
 	}
 }

+// StateRm runs terraform state rm for the project described by ctx.
+func (p *DefaultProjectCommandRunner) StateRm(ctx command.ProjectContext) command.ProjectResult {
+	stateRmSuccess, failure, err := p.doStateRm(ctx)
+	return command.ProjectResult{
+		Command:        command.State,
+		SubCommand:     "rm",
+		StateRmSuccess: stateRmSuccess,
+		Error:          err,
+		Failure:        failure,
+		RepoRelDir:     ctx.RepoRelDir,
+		Workspace:      ctx.Workspace,
+		ProjectName:    ctx.ProjectName,
+	}
+}
+
 func (p *DefaultProjectCommandRunner) doApprovePolicies(ctx command.ProjectContext) (*models.PolicyCheckSuccess, string, error) {

 	// TODO: Make this a bit smarter
@@ -539,6 +561,47 @@ func (p *DefaultProjectCommandRunner) doImport(ctx command.ProjectContext) (out
 	}, "", nil
 }

+func (p *DefaultProjectCommandRunner) doStateRm(ctx command.ProjectContext) (out *models.StateRmSuccess, failure string, err error) {
+	// Clone is idempotent so okay to run even if the repo was already cloned.
+	repoDir, _, cloneErr := p.WorkingDir.Clone(ctx.Log, ctx.HeadRepo, ctx.Pull, ctx.Workspace)
+	if cloneErr != nil {
+		return nil, "", cloneErr
+	}
+	projAbsPath := filepath.Join(repoDir, ctx.RepoRelDir)
+	if _, err = os.Stat(projAbsPath); os.IsNotExist(err) {
+		return nil, "", DirNotExistErr{RepoRelDir: ctx.RepoRelDir}
+	}
+
+	// Acquire Atlantis lock for this repo/dir/workspace.
+	lockAttempt, err := p.Locker.TryLock(ctx.Log, ctx.Pull, ctx.User, ctx.Workspace, models.NewProject(ctx.Pull.BaseRepo.FullName, ctx.RepoRelDir), ctx.RepoLocking)
+	if err != nil {
+		return nil, "", errors.Wrap(err, "acquiring lock")
+	}
+	if !lockAttempt.LockAcquired {
+		return nil, lockAttempt.LockFailureReason, nil
+	}
+	ctx.Log.Debug("acquired lock for project")
+
+	// Acquire internal lock for the directory we're going to operate in.
+	unlockFn, err := p.WorkingDirLocker.TryLock(ctx.Pull.BaseRepo.FullName, ctx.Pull.Num, ctx.Workspace, ctx.RepoRelDir)
+	if err != nil {
+		return nil, "", err
+	}
+	defer unlockFn()
+
+	outputs, err := p.runSteps(ctx.Steps, ctx, projAbsPath)
+	if err != nil {
+		return nil, "", fmt.Errorf("%s\n%s", err, strings.Join(outputs, "\n"))
+	}
+
+	// after state rm, re-plan command is required without state rm args
+	rePlanCmd := strings.TrimSpace(strings.Split(ctx.RePlanCmd, "--")[0])
+	return &models.StateRmSuccess{
+		Output:    strings.Join(outputs, "\n"),
+		RePlanCmd: rePlanCmd,
+	}, "", nil
+}
+
 func (p *DefaultProjectCommandRunner) runSteps(steps []valid.Step, ctx command.ProjectContext, absPath string) ([]string, error) {
 	var outputs []string

@@ -561,6 +624,8 @@ func (p *DefaultProjectCommandRunner) runSteps(steps []valid.Step, ctx command.P
 			out, err = p.VersionStepRunner.Run(ctx, step.ExtraArgs, absPath, envs)
 		case "import":
 			out, err = p.ImportStepRunner.Run(ctx, step.ExtraArgs, absPath, envs)
+		case "state_rm":
+			out, err = p.StateRmStepRunner.Run(ctx, step.ExtraArgs, absPath, envs)
 		case "run":
 			out, err = p.RunStepRunner.Run(ctx, step.RunCommand, absPath, envs, true)
 		case "env":
--- a/server/events/project_command_runner_test.go
+++ b/server/events/project_command_runner_test.go
@@ -684,6 +684,7 @@ func TestDefaultProjectCommandRunner_Import(t *testing.T) {
 			RegisterMockTestingT(t)
 			mockInit := mocks.NewMockStepRunner()
 			mockImport := mocks.NewMockStepRunner()
+			mockStateRm := mocks.NewMockStepRunner()
 			mockWorkingDir := mocks.NewMockWorkingDir()
 			mockLocker := mocks.NewMockProjectLocker()
 			mockSender := mocks.NewMockWebhooksSender()
@@ -696,6 +697,7 @@ func TestDefaultProjectCommandRunner_Import(t *testing.T) {
 				LockURLGenerator:          mockURLGenerator{},
 				InitStepRunner:            mockInit,
 				ImportStepRunner:          mockImport,
+				StateRmStepRunner:         mockStateRm,
 				WorkingDir:                mockWorkingDir,
 				Webhooks:                  mockSender,
 				WorkingDirLocker:          events.NewDefaultWorkingDirLocker(),
--- a/server/events/pull_updater.go
+++ b/server/events/pull_updater.go
@@ -28,7 +28,7 @@ func (c *PullUpdater) updatePull(ctx *command.Context, cmd PullCommand, res comm
 		}
 	}

-	comment := c.MarkdownRenderer.Render(res, cmd.CommandName(), ctx.Log.GetHistory(), cmd.IsVerbose(), ctx.Pull.BaseRepo.VCSHost.Type)
+	comment := c.MarkdownRenderer.Render(res, cmd.CommandName(), cmd.SubCommandName(), ctx.Log.GetHistory(), cmd.IsVerbose(), ctx.Pull.BaseRepo.VCSHost.Type)
 	if err := c.VCSClient.CreateComment(ctx.Pull.BaseRepo, ctx.Pull.Num, comment, cmd.CommandName().String()); err != nil {
 		ctx.Log.Err("unable to comment: %s", err)
 	}
--- a/server/events/state_command_runner.go
+++ b/server/events/state_command_runner.go
@@ -0,0 +1,46 @@
+package events
+
+import (
+	"fmt"
+
+	"github.com/runatlantis/atlantis/server/events/command"
+)
+
+func NewStateCommandRunner(
+	pullUpdater *PullUpdater,
+	prjCmdBuilder ProjectStateCommandBuilder,
+	prjCmdRunner ProjectStateCommandRunner,
+) *StateCommandRunner {
+	return &StateCommandRunner{
+		pullUpdater:   pullUpdater,
+		prjCmdBuilder: prjCmdBuilder,
+		prjCmdRunner:  prjCmdRunner,
+	}
+}
+
+type StateCommandRunner struct {
+	pullUpdater   *PullUpdater
+	prjCmdBuilder ProjectStateCommandBuilder
+	prjCmdRunner  ProjectStateCommandRunner
+}
+
+func (v *StateCommandRunner) Run(ctx *command.Context, cmd *CommentCommand) {
+	var result command.Result
+	switch cmd.SubName {
+	case "rm":
+		result = v.runRm(ctx, cmd)
+	default:
+		result = command.Result{
+			Failure: fmt.Sprintf("unknown state subcommand %s", cmd.SubName),
+		}
+	}
+	v.pullUpdater.updatePull(ctx, cmd, result)
+}
+
+func (v *StateCommandRunner) runRm(ctx *command.Context, cmd *CommentCommand) command.Result {
+	projectCmds, err := v.prjCmdBuilder.BuildStateRmCommands(ctx, cmd)
+	if err != nil {
+		ctx.Log.Warn("Error %s", err)
+	}
+	return runProjectCmds(projectCmds, v.prjCmdRunner.StateRm)
+}
--- a/server/events/templates/import_success_unwrapped.tmpl
+++ b/server/events/templates/import_success_unwrapped.tmpl
@@ -3,6 +3,8 @@
 {{ .Output }}
 ```

+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
-  * `{{ .RePlanCmd }}`
+  * `{{.RePlanCmd}}`
 {{ end -}}
--- a/server/events/templates/import_success_wrapped.tmpl
+++ b/server/events/templates/import_success_wrapped.tmpl
@@ -4,6 +4,8 @@
 {{ .Output }}
 ```
 </details>
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
 * :repeat: To **plan** this project again, comment:
  * `{{ .RePlanCmd }}`
 {{ end -}}
--- a/server/events/templates/multi_project_state_rm.tmpl
+++ b/server/events/templates/multi_project_state_rm.tmpl
@@ -0,0 +1,10 @@
+{{ define "multiProjectStateRm" -}}
+{{ template "multiProjectHeader" . }}
+{{ range $i, $result := .Results -}}
+### {{ add $i 1 }}. {{ if $result.ProjectName }}project: `{{ $result.ProjectName }}` {{ end }}dir: `{{ $result.RepoRelDir }}` workspace: `{{ $result.Workspace }}`
+{{ $result.Rendered}}
+
+---
+{{ end -}}
+{{- template "log" . -}}
+{{ end -}}
--- a/server/events/templates/single_project_state_rm_success.tmpl
+++ b/server/events/templates/single_project_state_rm_success.tmpl
@@ -0,0 +1,6 @@
+{{ define "singleProjectStateRm" -}}
+{{$result := index .Results 0}}Ran {{.Command}} `{{.SubCommand}}` for {{ if $result.ProjectName }}project: `{{$result.ProjectName}}` {{ end }}dir: `{{$result.RepoRelDir}}` workspace: `{{$result.Workspace}}`
+
+{{$result.Rendered}}
+{{ template "log" . }}
+{{ end }}
--- a/server/events/templates/state_rm_success_unwrapped.tmpl
+++ b/server/events/templates/state_rm_success_unwrapped.tmpl
@@ -0,0 +1,10 @@
+{{ define "stateRmSuccessUnwrapped" -}}
+```diff
+{{ .Output }}
+```
+
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `{{.RePlanCmd}}`
+{{ end }}
--- a/server/events/templates/state_rm_success_wrapped.tmpl
+++ b/server/events/templates/state_rm_success_wrapped.tmpl
@@ -0,0 +1,11 @@
+{{ define "stateRmSuccessWrapped" -}}
+<details><summary>Show Output</summary>
+```diff
+{{ .Output }}
+```
+</details>
+:put_litter_in_its_place: A plan file was discarded. Re-plan would be required before applying.
+
+* :repeat: To **plan** this project again, comment:
+  * `{{.RePlanCmd}}`
+{{ end }}
--- a/server/server.go
+++ b/server/server.go
@@ -619,6 +619,7 @@ func NewServer(userConfig UserConfig, config Config) (*Server, error) {
 			DefaultTFVersion:  defaultTfVersion,
 		},
 		ImportStepRunner:          runtime.NewImportStepRunner(terraformClient, defaultTfVersion),
+		StateRmStepRunner:         runtime.NewStateRmStepRunner(terraformClient, defaultTfVersion),
 		WorkingDir:                workingDir,
 		Webhooks:                  webhooksManager,
 		WorkingDirLocker:          workingDirLocker,
@@ -730,6 +731,12 @@ func NewServer(userConfig UserConfig, config Config) (*Server, error) {
 		instrumentedProjectCmdRunner,
 	)

+	stateCommandRunner := events.NewStateCommandRunner(
+		pullUpdater,
+		projectCommandBuilder,
+		instrumentedProjectCmdRunner,
+	)
+
 	commentCommandRunnerByCmd := map[command.Name]events.CommentCommandRunner{
 		command.Plan:            planCommandRunner,
 		command.Apply:           applyCommandRunner,
@@ -737,6 +744,7 @@ func NewServer(userConfig UserConfig, config Config) (*Server, error) {
 		command.Unlock:          unlockCommandRunner,
 		command.Version:         versionCommandRunner,
 		command.Import:          importCommandRunner,
+		command.State:           stateCommandRunner,
 	}

 	githubTeamAllowlistChecker, err := events.NewTeamAllowlistChecker(userConfig.GithubTeamAllowlist)
--- a/server/user_config_test.go
+++ b/server/user_config_test.go
@@ -19,23 +19,23 @@ func TestUserConfig_ToAllowCommandNames(t *testing.T) {
 	}{
 		{
 			name:          "full commands can be parsed by comma",
-			allowCommands: "apply,plan,unlock,policy_check,approve_policies,version,import",
+			allowCommands: "apply,plan,unlock,policy_check,approve_policies,version,import,state",
 			want: []command.Name{
-				command.Apply, command.Plan, command.Unlock, command.PolicyCheck, command.ApprovePolicies, command.Version, command.Import,
+				command.Apply, command.Plan, command.Unlock, command.PolicyCheck, command.ApprovePolicies, command.Version, command.Import, command.State,
 			},
 		},
 		{
 			name:          "all",
 			allowCommands: "all",
 			want: []command.Name{
-				command.Version, command.Plan, command.Apply, command.Unlock, command.ApprovePolicies, command.Import,
+				command.Version, command.Plan, command.Apply, command.Unlock, command.ApprovePolicies, command.Import, command.State,
 			},
 		},
 		{
 			name:          "all with others returns same with all result",
 			allowCommands: "all,plan",
 			want: []command.Name{
-				command.Version, command.Plan, command.Apply, command.Unlock, command.ApprovePolicies, command.Import,
+				command.Version, command.Plan, command.Apply, command.Unlock, command.ApprovePolicies, command.Import, command.State,
 			},
 		},
 		{